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.
+
+*****************
+ Output commands
+*****************
+
+The shared router supplies one device FIFO for singles and batches. The
+driver checks ownership, output access, value type, and expiry before
+physical work. Steady outputs accept Boolean values. Pulse outputs
+accept only true and complete after the function 05 acknowledgement,
+without coil readback or transient reports. Analog outputs accept JSON
+numbers from zero through ten volts and round them to millivolts. Exact
+range checks precede floating-point conversion.
+
+A batch accepts steady digital outputs only. It sorts eligible targets
+by coil address, writes each contiguous range with function 15, and uses
+function 05 for an isolated target. Gaps remain unchanged. It reads the
+complete coil range once after all writes. A transport failure fails all
+pending members, including those in an earlier successful range. A
+mismatch fails only that member. Commands that expire before their write
+are rejected. Physical writes also have the earliest member deadline.
+
+Changed feedback is sent before results. Each changed commanded point
+has its command ID as causation. Unrelated output changes have no
+command cause. Equal feedback needs no second report. The framework
+ledger records all outcomes before feedback or result publication, so a
+failed send cannot cause another physical action on replay. A repeated
+command reuses its first result ID, time, status, and reason. The ledger
+is memory-only and bounded to 256 records for the assembly. An eviction
+or process restart removes that duplicate protection.
+
+Commands received during device backoff are rejected if invalid or
+expired, and fail as unavailable otherwise. No command executes before
+the connection setup and baseline complete. Shutdown does not retry a
+physical action. Results use fixed failure reasons without transport
+addresses or untrusted response data. Drivers return errors and do not
+log them. Publication and independent cleanup failures stop the driver
+through the runner.
+
+Fake tests cover single writes, cross-bank batches, gaps, partial
+transport failure, mismatches, pulses, analog boundaries, expired work,
+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.
--- /dev/null
+package m31
+
+import (
+ "cmp"
+ "context"
+ "errors"
+ "fmt"
+ "slices"
+ "time"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+type commandHandler struct {
+ connection *connection
+}
+
+func (h *commandHandler) handle(ctx context.Context, commands []driver.Command) error {
+ pending, err := h.prepare(ctx, commands)
+ if err != nil {
+ return fmt.Errorf("prepare output commands: %w", err)
+ }
+
+ if len(pending) == 0 {
+ return nil
+ }
+
+ if len(commands) == 1 && (pending[0].analog || pending[0].point.pulse != 0) {
+ err = h.special(ctx, pending[0])
+ if err != nil {
+ return fmt.Errorf("apply special output: %w", err)
+ }
+
+ return nil
+ }
+
+ slices.SortFunc(pending, func(first outputRequest, second outputRequest) int {
+ return cmp.Compare(first.point.address, second.point.address)
+ })
+
+ active := make([]outputRequest, 0, len(pending))
+
+ for start := 0; start < len(pending); {
+ if pending[start].command.Expired(time.Now()) {
+ err = h.result(ctx, pending[start].command, protocol.ResultPayload{
+ Status: "rejected",
+ Reason: "Command expired",
+ })
+ if err != nil {
+ return fmt.Errorf("reject expired write: %w", err)
+ }
+
+ start++
+
+ continue
+ }
+
+ end := start + 1
+
+ for end < len(pending) && pending[end].point.address == pending[end-1].point.address+1 && !pending[end].command.Expired(time.Now()) {
+ end++
+ }
+
+ err = h.writeRange(ctx, pending[start:end])
+ if err != nil {
+ return fmt.Errorf("fail coil writes: %w", h.fail(ctx, pending, err))
+ }
+
+ active = append(active, pending[start:end]...)
+ start = end
+ }
+
+ if len(active) == 0 {
+ return nil
+ }
+
+ next := h.connection.state
+ next.outputs, err = h.connection.owner.client.ReadCoils(ctx, uint16(len(h.connection.owner.topology.outputs)))
+ if err != nil {
+ return fmt.Errorf("fail coil readback: %w", h.fail(ctx, active, fmt.Errorf("read command coils: %w", err)))
+ }
+
+ if len(next.outputs) != len(h.connection.owner.topology.outputs) {
+ return fmt.Errorf("fail incomplete readback: %w", h.fail(ctx, active, fmt.Errorf("incomplete coil readback")))
+ }
+
+ next.observedAt = time.Now().UTC()
+
+ err = h.complete(ctx, active, next)
+ if err != nil {
+ return fmt.Errorf("complete coil commands: %w", err)
+ }
+
+ return nil
+}
+
+func (h *commandHandler) prepare(ctx context.Context, commands []driver.Command) ([]outputRequest, error) {
+ var pending []outputRequest
+
+ for _, command := range commands {
+ record, exists := h.connection.owner.ledger.Lookup(command.Metadata.MessageID)
+ if exists {
+ err := h.send(ctx, command.Metadata.MessageID, record)
+ if err != nil {
+ return nil, fmt.Errorf("replay result: %w", err)
+ }
+
+ continue
+ }
+
+ request, reason := newOutputRequest(command, h.connection.owner.topology, h.connection.owner.deviceID, len(commands) > 1)
+ if reason != "" {
+ err := h.result(ctx, command, protocol.ResultPayload{
+ Status: "rejected",
+ Reason: reason,
+ })
+ if err != nil {
+ return nil, fmt.Errorf("reject command: %w", err)
+ }
+
+ continue
+ }
+
+ pending = append(pending, request)
+ }
+
+ return pending, nil
+}
+
+func (h *commandHandler) send(ctx context.Context, commandID string, record driver.ResultRecord) error {
+ err := h.connection.owner.publisher.Result(ctx, h.connection.owner.deviceID, commandID, record.Metadata, record.Payload)
+ if err != nil {
+ return fmt.Errorf("publish result: %w", err)
+ }
+
+ return nil
+}
+
+func (h *commandHandler) result(ctx context.Context, command driver.Command, payload protocol.ResultPayload) error {
+ record, err := h.connection.owner.ledger.Remember(command.Metadata.MessageID, payload)
+ if err != nil {
+ return fmt.Errorf("remember command result: %w", err)
+ }
+
+ err = h.send(ctx, command.Metadata.MessageID, record)
+ if err != nil {
+ return fmt.Errorf("send command result: %w", err)
+ }
+
+ return nil
+}
+
+func (h *commandHandler) writeRange(ctx context.Context, requests []outputRequest) error {
+ expires := *requests[0].command.Metadata.ExpiresAt
+ values := make([]bool, len(requests))
+
+ for index, request := range requests {
+ values[index] = request.value == 1
+ if request.command.Metadata.ExpiresAt.Before(expires) {
+ expires = *request.command.Metadata.ExpiresAt
+ }
+ }
+
+ ctx, cancel := context.WithDeadline(ctx, expires)
+ defer cancel()
+
+ var err error
+
+ if len(requests) == 1 {
+ err = h.connection.owner.client.WriteCoil(ctx, requests[0].point.address, values[0])
+ } else {
+ err = h.connection.owner.client.WriteCoils(ctx, requests[0].point.address, values)
+ }
+
+ if err != nil {
+ return fmt.Errorf("write output coils: %w", err)
+ }
+
+ return nil
+}
+
+func (h *commandHandler) fail(ctx context.Context, requests []outputRequest, cause error) error {
+ records := make([]driver.ResultRecord, len(requests))
+
+ for index, request := range requests {
+ record, err := h.connection.owner.ledger.Remember(request.command.Metadata.MessageID, protocol.ResultPayload{
+ Status: "failed",
+ Reason: "Modbus operation failed",
+ })
+ if err != nil {
+ return fmt.Errorf("remember failed commands: %w", errors.Join(cause, err))
+ }
+
+ records[index] = record
+ }
+
+ for index, request := range requests {
+ err := h.send(ctx, request.command.Metadata.MessageID, records[index])
+ if err != nil {
+ return fmt.Errorf("send failed commands: %w", errors.Join(cause, err))
+ }
+ }
+
+ return &deviceError{err: cause}
+}
+
+func (h *commandHandler) complete(ctx context.Context, requests []outputRequest, next snapshot) error {
+ records := make([]driver.ResultRecord, len(requests))
+ causes := make(map[string]string, len(requests))
+
+ for index, request := range requests {
+ actual := uint16(0)
+ if request.analog {
+ actual = next.analog[request.point.address]
+ } else if next.outputs[request.point.address] {
+ actual = 1
+ }
+
+ payload := protocol.ResultPayload{Status: "completed"}
+ if actual != request.value {
+ payload.Status, payload.Reason = "failed", "Output readback differs from command"
+ }
+
+ record, err := h.connection.owner.ledger.Remember(request.command.Metadata.MessageID, payload)
+ if err != nil {
+ return fmt.Errorf("remember output results: %w", err)
+ }
+
+ records[index] = record
+ causes[request.point.id] = request.command.Metadata.MessageID
+ }
+
+ err := h.connection.publish(ctx, next, false, causes)
+ if err != nil {
+ return fmt.Errorf("publish output feedback: %w", err)
+ }
+
+ for index, request := range requests {
+ err = h.send(ctx, request.command.Metadata.MessageID, records[index])
+ if err != nil {
+ return fmt.Errorf("publish completed results: %w", err)
+ }
+ }
+
+ return nil
+}
+
+func (h *commandHandler) special(ctx context.Context, request outputRequest) error {
+ next, err := h.applySpecial(ctx, request)
+ if err != nil {
+ return fmt.Errorf("fail special command: %w", h.fail(ctx, []outputRequest{request}, err))
+ }
+
+ if !request.analog {
+ err = h.result(ctx, request.command, protocol.ResultPayload{Status: "completed"})
+ if err != nil {
+ return fmt.Errorf("complete pulse: %w", err)
+ }
+
+ return nil
+ }
+
+ err = h.complete(ctx, []outputRequest{request}, next)
+ if err != nil {
+ return fmt.Errorf("complete voltage command: %w", err)
+ }
+
+ return nil
+}
+
+func (h *commandHandler) applySpecial(ctx context.Context, request outputRequest) (snapshot, error) {
+ ctx, cancel := context.WithDeadline(ctx, *request.command.Metadata.ExpiresAt)
+ defer cancel()
+
+ if !request.analog {
+ err := h.connection.owner.client.WriteCoil(ctx, request.point.address, true)
+ if err != nil {
+ return snapshot{}, fmt.Errorf("write pulse: %w", err)
+ }
+
+ return h.connection.state, nil
+ }
+
+ err := h.connection.owner.client.WriteRegister(ctx, request.point.address, request.value)
+ if err != nil {
+ return snapshot{}, fmt.Errorf("write voltage: %w", err)
+ }
+
+ next := h.connection.state
+ next.analog, err = h.connection.owner.client.ReadRegisters(ctx, 0, uint16(len(h.connection.owner.topology.analog)))
+ if err != nil {
+ return snapshot{}, fmt.Errorf("read voltage feedback: %w", err)
+ }
+
+ if len(next.analog) != len(h.connection.owner.topology.analog) {
+ return snapshot{}, fmt.Errorf("incomplete voltage feedback")
+ }
+
+ err = validateVoltage(next.analog)
+ if err != nil {
+ return snapshot{}, fmt.Errorf("validate voltage feedback: %w", err)
+ }
+
+ next.observedAt = time.Now().UTC()
+
+ return next, nil
+}
+
+func (h *commandHandler) unavailable(ctx context.Context, commands []driver.Command) error {
+ pending, err := h.prepare(ctx, commands)
+ if err != nil {
+ return fmt.Errorf("prepare unavailable commands: %w", err)
+ }
+
+ for _, request := range pending {
+ err = h.result(ctx, request.command, protocol.ResultPayload{
+ Status: "failed",
+ Reason: "Modbus device is unavailable",
+ })
+ if err != nil {
+ return fmt.Errorf("publish unavailable result: %w", err)
+ }
+ }
+
+ return nil
+}
--- /dev/null
+package m31
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "slices"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+func TestCommandsWriteContiguousRangesAndReplayResults(t *testing.T) {
+ c, client, transport := testConnection(t)
+ h := &commandHandler{connection: c}
+
+ var commands []driver.Command
+
+ for address := 5; address < 18; address++ {
+ commands = append(commands, testCommand(t, c.owner.topology.outputs[address].id, "true"))
+ }
+
+ slices.Reverse(commands)
+ require.NoError(t, h.handle(t.Context(), commands))
+ require.Len(t, client.writes, 1)
+ require.EqualValues(t, 5, client.writes[0].address)
+ require.True(t, client.writes[0].multiple)
+ require.Len(t, client.writes[0].values, 13)
+ require.Equal(t, 1, client.coilReads)
+
+ reports := transport.ofKind(protocol.Reported)
+ results := transport.ofKind(protocol.Result)
+
+ require.Len(t, reports, 13)
+ require.Len(t, results, 13)
+
+ for _, command := range commands {
+ for _, report := range reports {
+ if report.Address.Target == command.Capability {
+ require.Equal(t, command.Metadata.MessageID, report.Metadata.CausationID)
+ }
+ }
+ }
+
+ require.NoError(t, h.handle(t.Context(), commands))
+ require.Len(t, client.writes, 1)
+ require.Equal(t, 1, client.coilReads)
+
+ replayed := transport.ofKind(protocol.Result)[13:]
+ byCommand := make(map[string]protocol.Metadata)
+
+ for _, result := range results {
+ byCommand[result.Address.Target] = result.Metadata
+ }
+
+ for _, result := range replayed {
+ require.Equal(t, byCommand[result.Address.Target], result.Metadata)
+ }
+}
+
+func TestCommandsKeepGapsAndRejectNonsteadyMembers(t *testing.T) {
+ c, client, transport := testConnection(t)
+ h := &commandHandler{connection: c}
+ commands := []driver.Command{
+ testCommand(t, "m0.do0", "true"),
+ testCommand(t, "m0.do1", "true"),
+ testCommand(t, "m0.do2", "true"),
+ testCommand(t, "m0.do4", "true"),
+ testCommand(t, "m2.ao0", "2"),
+ testCommand(t, "m0.di0", "true"),
+ }
+ require.NoError(t, h.handle(t.Context(), commands))
+ require.Len(t, client.writes, 2)
+ require.EqualValues(t, 0, client.writes[0].address)
+ require.Len(t, client.writes[0].values, 2)
+ require.EqualValues(t, 4, client.writes[1].address)
+ require.False(t, client.writes[1].multiple)
+ require.False(t, client.coils[2])
+ require.False(t, client.coils[3])
+ require.Len(t, transport.ofKind(protocol.Result), 6)
+}
+
+func TestCommandsFailAllOnTransportErrorAndOnlyMismatchOtherwise(t *testing.T) {
+ testCases := []struct {
+ name string
+ change func(*fakeClient)
+ transportFailure bool
+ }{
+ {name: "fails all on write failure", transportFailure: true, change: func(c *fakeClient) {
+ c.writeErr = errors.New("write failed")
+ }},
+ {name: "fails all after a partial write", transportFailure: true, change: func(c *fakeClient) {
+ c.writeErr = errors.New("later write failed")
+ c.failAfterWrites = 1
+ }},
+ {name: "fails all on read failure", transportFailure: true, change: func(c *fakeClient) {
+ c.readErr = errors.New("read failed")
+ }},
+ {name: "fails all on short read", transportFailure: true, change: func(c *fakeClient) {
+ c.shortRead = true
+ }},
+ {name: "fails only mismatching member", change: func(c *fakeClient) {
+ c.mismatch = 1
+ }},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ c, client, transport := testConnection(t)
+ tc.change(client)
+
+ h := &commandHandler{connection: c}
+ commands := []driver.Command{testCommand(t, "m0.do0", "true"), testCommand(t, "m0.do1", "true"), testCommand(t, "m1.do7", "true")}
+ err := h.handle(t.Context(), commands)
+ if tc.transportFailure {
+ var failure *deviceError
+
+ require.ErrorAs(t, err, &failure)
+ } else {
+ require.NoError(t, err)
+ }
+
+ require.Len(t, transport.ofKind(protocol.Result), 3)
+
+ for index, command := range commands {
+ record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+ require.True(t, exists)
+
+ status := "completed"
+ if tc.transportFailure || index == 1 {
+ status = "failed"
+ }
+
+ require.Equal(t, status, record.Payload.Status)
+ }
+ })
+ }
+}
+
+func TestCommandsCompletePulseWithoutReadbackOrFeedback(t *testing.T) {
+ c, client, transport := testConnection(t)
+ h := &commandHandler{connection: c}
+ command := testCommand(t, "m0.do2", "true")
+ require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+ require.Len(t, client.writes, 1)
+ require.Zero(t, client.coilReads)
+ require.Empty(t, transport.ofKind(protocol.Reported))
+ require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+ require.Len(t, client.writes, 1)
+
+ command = testCommand(t, "m0.do2", "false")
+ require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+ require.Len(t, client.writes, 1)
+
+ record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+ require.True(t, exists)
+ require.Equal(t, "rejected", record.Payload.Status)
+}
+
+func TestCommandsWriteVoltageAndCheckFeedback(t *testing.T) {
+ testCases := []struct {
+ name string
+ change func(*fakeClient)
+ wantStatus string
+ wantError bool
+ }{
+ {name: "completes verified voltage", wantStatus: "completed", change: func(*fakeClient) {
+ }},
+ {name: "fails mismatching voltage", wantStatus: "failed", change: func(c *fakeClient) {
+ c.mismatch = 3
+ }},
+ {name: "fails register transport", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+ c.registerErr = errors.New("register failed")
+ }},
+ {name: "fails analog read transport", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+ c.registerReadErr = errors.New("read failed")
+ }},
+ {name: "fails unsafe feedback", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+ c.analog[0] = 10001
+ }},
+ {name: "fails incomplete feedback", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+ c.analog = c.analog[:4]
+ }},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ c, client, transport := testConnection(t)
+ tc.change(client)
+
+ h := &commandHandler{connection: c}
+ command := testCommand(t, "m2.ao3", "5.5")
+ err := h.handle(t.Context(), []driver.Command{command})
+ if tc.wantError {
+ require.Error(t, err)
+ } else {
+ require.NoError(t, err)
+ }
+
+ record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+ require.True(t, exists)
+ require.Equal(t, tc.wantStatus, record.Payload.Status)
+
+ if tc.wantStatus == "completed" {
+ require.EqualValues(t, 5500, client.analog[3])
+ require.Len(t, transport.messages, 2)
+ require.Equal(t, protocol.Reported, transport.messages[0].Address.Kind)
+ require.Equal(t, command.Metadata.MessageID, transport.messages[0].Metadata.CausationID)
+ }
+ })
+ }
+}
+
+func TestCommandsRememberBeforeFailedPublication(t *testing.T) {
+ for _, kind := range []protocol.Kind{protocol.Reported, protocol.Result} {
+ t.Run(fmt.Sprint("preserves first outcome after ", kind, " failure"), func(t *testing.T) {
+ c, client, transport := testConnection(t)
+ h := &commandHandler{connection: c}
+ failure := errors.New("publication failed")
+ transport.failKind, transport.err = kind, failure
+ commands := []driver.Command{testCommand(t, "m0.do0", "true"), testCommand(t, "m0.do1", "true")}
+ require.ErrorIs(t, h.handle(t.Context(), commands), failure)
+
+ record, exists := c.owner.ledger.Lookup(commands[1].Metadata.MessageID)
+ require.True(t, exists)
+ require.Equal(t, "completed", record.Payload.Status)
+
+ transport.err = nil
+
+ require.NoError(t, h.handle(t.Context(), commands))
+ require.Len(t, client.writes, 1)
+ })
+ }
+}
+
+func TestCommandsRecheckExpiryBeforeEachWriteAndHonorCancellation(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ c, client, _ := testConnection(t)
+ h := &commandHandler{connection: c}
+ first := testCommand(t, "m0.do0", "true")
+ second := testCommand(t, "m0.do4", "true")
+ expires := time.Now().Add(time.Millisecond)
+ second.Metadata.ExpiresAt = &expires
+ client.writeDelay = 2 * time.Millisecond
+
+ require.NoError(t, h.handle(t.Context(), []driver.Command{first, second}))
+ require.Len(t, client.writes, 1)
+
+ record, exists := c.owner.ledger.Lookup(second.Metadata.MessageID)
+ require.True(t, exists)
+ require.Equal(t, "rejected", record.Payload.Status)
+
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+
+ third := testCommand(t, "m0.do1", "true")
+ require.ErrorIs(t, h.handle(ctx, []driver.Command{third}), context.Canceled)
+ require.Len(t, client.writes, 1)
+ })
+}
+
+func TestCommandsCompleteUnchangedSingleWithoutState(t *testing.T) {
+ c, client, transport := testConnection(t)
+ h := &commandHandler{connection: c}
+ command := testCommand(t, "m0.do0", "false")
+ require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+ require.Len(t, client.writes, 1)
+ require.False(t, client.writes[0].multiple)
+ require.Len(t, transport.messages, 1)
+ require.Equal(t, protocol.Result, transport.messages[0].Address.Kind)
+}
+
+func TestCommandsReturnPublicationErrorsWithoutLosingOutcomes(t *testing.T) {
+ testCases := []struct {
+ name string
+ capability string
+ value string
+ offline bool
+ physicalFailure bool
+ }{
+ {name: "retains completed pulse after result failure", capability: "m0.do2", value: "true"},
+ {name: "retains voltage after result failure", capability: "m2.ao0", value: "5"},
+ {name: "returns rejection failure", capability: "m0.di0", value: "true"},
+ {name: "returns offline result failure", capability: "m0.do0", value: "true", offline: true},
+ {name: "returns offline rejection failure", capability: "m0.di0", value: "true", offline: true},
+ {name: "joins pulse and result failure", capability: "m0.do2", value: "true", physicalFailure: true},
+ {name: "joins batch and result failure", capability: "m0.do0", value: "true", physicalFailure: true},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ c, client, transport := testConnection(t)
+ h := &commandHandler{connection: c}
+ command := testCommand(t, tc.capability, tc.value)
+ sendErr := errors.New("send failed")
+ physicalErr := errors.New("physical failure")
+ transport.failKind, transport.err = protocol.Result, sendErr
+ if tc.physicalFailure {
+ client.writeErr = physicalErr
+ }
+
+ var err error
+
+ if tc.offline {
+ err = h.unavailable(t.Context(), []driver.Command{command})
+ } else {
+ err = h.handle(t.Context(), []driver.Command{command})
+ }
+
+ require.ErrorIs(t, err, sendErr)
+
+ if tc.physicalFailure {
+ require.ErrorIs(t, err, physicalErr)
+
+ var physical *deviceError
+
+ require.NotErrorAs(t, err, &physical)
+ }
+
+ record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+ require.True(t, exists)
+
+ transport.err, client.writeErr = nil, nil
+ count := len(client.writes)
+
+ require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+ require.Len(t, client.writes, count)
+
+ replay := transport.ofKind(protocol.Result)
+ require.Equal(t, record.Metadata, replay[len(replay)-1].Metadata)
+ })
+ }
+}
{name: "rejects pulse register write failure", change: func(c *fakeClient) {
c.registerErr = errors.New("configuration failed")
}},
+ {name: "rejects configuration read failure", change: func(c *fakeClient) {
+ c.registerReadErr = errors.New("configuration read failed")
+ }},
+ {name: "rejects pulse width mismatch", change: func(c *fakeClient) {
+ c.wrongRegister = pulseWidthBase + 2
+ }},
{name: "rejects configuration mismatch", change: func(c *fakeClient) {
c.wrongConfig = true
}},
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- d, client, transport := testAssembly(t)
+ d, client, transport, _ := testAssembly(t)
tc.change(client)
c := &connection{owner: d}
"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.
+// Driver serializes one assembly's polling and commands. 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. Duplicate protection ends at ledger
+// eviction or process restart. No operation starts a physical connection before
+// Run.
type Driver struct {
deviceID string
address string
topology topology
publisher *driver.Publisher
client Client
+ queue <-chan []driver.Command
+ ledger *driver.Ledger
}
-// 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) {
+// New validates settings and requires the device's queue in router. The caller
+// supplies valid publisher, router, and client dependencies and starts the
+// router's intake separately. The driver owns the client until Run returns.
+func New(settings Settings, publisher *driver.Publisher, router *driver.Router, client Client) (*Driver, error) {
if !protocol.ValidID(settings.DeviceID) {
return nil, fmt.Errorf("device ID must be a canonical ULID")
}
return nil, fmt.Errorf("validate topology: %w", err)
}
+ queue, exists := router.Queue(settings.DeviceID)
+ if !exists {
+ return nil, fmt.Errorf("device has no command queue")
+ }
+
+ ledger, err := driver.NewLedger(settings.DeviceID)
+ if err != nil {
+ return nil, fmt.Errorf("create result ledger: %w", err)
+ }
+
return &Driver{
deviceID: settings.DeviceID,
address: settings.Address,
topology: points,
publisher: publisher,
client: client,
+ queue: queue,
+ ledger: ledger,
}, nil
}
-// Run configures and verifies output modes before online or polling. A
+// Run configures and verifies output modes before online or command work. 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.
+// baseline succeeds. Commands received during backoff fail without I/O.
func (d *Driver) Run(ctx context.Context) error {
err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
if err != nil {
}
healthy = true
+ handler := &commandHandler{connection: connection}
ticker := time.NewTicker(d.pollInterval)
defer ticker.Stop()
select {
case <-ctx.Done():
return healthy, false, ctx.Err()
+ case commands := <-d.queue:
+ err = handler.handle(ctx, commands)
+ if err != nil {
+ return healthy, false, fmt.Errorf("process commands: %w", err)
+ }
case <-ticker.C:
err = connection.poll(ctx)
if err != nil {
timer := time.NewTimer(delay)
defer timer.Stop()
- select {
- case <-ctx.Done():
- return ctx.Err()
- case <-timer.C:
- return nil
+ handler := &commandHandler{connection: &connection{owner: d}}
+
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-timer.C:
+ return nil
+ case commands := <-d.queue:
+ err := handler.unavailable(ctx, commands)
+ if err != nil {
+ return fmt.Errorf("finish unavailable commands: %w", err)
+ }
+ }
}
}
"github.com/stretchr/testify/require"
+ "onehouse/driver"
"onehouse/protocol"
)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- d, client, _ := testAssembly(t)
+ d, client, _, router := testAssembly(t)
settings := Settings{
DeviceID: testDeviceID,
Address: "example.com:502",
}
tc.change(&settings)
- _, err := New(settings, d.publisher, client)
+ _, err := New(settings, d.publisher, router, client)
require.Error(t, err)
})
}
+
+ d, client, _, _ := testAssembly(t)
+ router, err := driver.NewRouter(nil)
+ require.NoError(t, err)
+
+ _, err = New(Settings{
+ DeviceID: testDeviceID,
+ Address: "example.com:502",
+ Unit: 1,
+ PollInterval: 50 * time.Millisecond,
+ Modules: []Module{{Model: HostModel}},
+ }, d.publisher, router, client)
+ require.ErrorContains(t, err, "command queue")
}
func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
- d, client, transport := testAssembly(t)
+ d, client, transport, router := testAssembly(t)
client.connectErrors = []error{errors.New("unavailable")}
client.inputs[1] = true
ctx, cancel := context.WithCancel(t.Context())
})
require.Empty(t, transport.ofKind(protocol.Event))
+ command := testCommand(t, "m0.do0", "true")
+ queueCommands(t, router, []driver.Command{command})
+ synctest.Wait()
+
+ record, exists := d.ledger.Lookup(command.Metadata.MessageID)
+ require.True(t, exists)
+ require.Equal(t, "failed", record.Payload.Status)
+ client.withLock(func(client *fakeClient) {
+ require.Empty(t, client.writes)
+ })
time.Sleep(time.Second)
synctest.Wait()
client.withLock(func(client *fakeClient) {
require.Len(t, client.registerWrites, 66)
})
require.Len(t, transport.ofKind(protocol.Event), 1)
+ queueCommands(t, router, []driver.Command{command})
+ synctest.Wait()
+ client.withLock(func(client *fakeClient) {
+ require.Empty(t, client.writes)
+ })
+
+ command = testCommand(t, "m1.do15", "true")
+ queueCommands(t, router, []driver.Command{command})
+ synctest.Wait()
+ client.withLock(func(client *fakeClient) {
+ require.Len(t, client.writes, 1)
+ })
cancel()
require.ErrorIs(t, <-done, context.Canceled)
client.withLock(func(client *fakeClient) {
func TestDriverBoundsReconnectBackoff(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
- d, client, _ := testAssembly(t)
+ d, client, _, _ := testAssembly(t)
for range 8 {
client.withLock(func(client *fakeClient) {
client.connectErrors = append(client.connectErrors, errors.New("offline"))
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- d, client, transport := testAssembly(t)
+ d, client, transport, _ := testAssembly(t)
failure := errors.New("independent failure")
if tc.closeFailure {
client.closeErr = failure
func TestDriverRefreshDoesNotRepeatCommandsOrEvents(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
- d, client, transport := testAssembly(t)
+ d, client, transport, _ := testAssembly(t)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
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
+ registerReadErr error
+ wrongRegister uint16
+ failAfterWrites int
+ 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()
- if c.registerErr != nil || ctx.Err() != nil {
- return nil, errors.Join(c.registerErr, ctx.Err())
+ if c.registerErr != nil || c.registerReadErr != nil || ctx.Err() != nil {
+ return nil, errors.Join(c.registerErr, c.registerReadErr, ctx.Err())
}
if address >= pulseWidthBase {
value := c.registers[address]
- if c.wrongConfig {
+ if c.wrongConfig || address == c.wrongRegister {
value++
}
time.Sleep(c.writeDelay)
- if c.writeErr != nil || ctx.Err() != nil {
+ if c.writeErr != nil && len(c.writes) >= c.failAfterWrites || ctx.Err() != nil {
return errors.Join(c.writeErr, ctx.Err())
}
return slices.Clone(f.messages)
}
-func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport) {
+func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport, *driver.Router) {
t.Helper()
transport := &fakeTransport{}
publisher := driver.NewPublisher(transport)
+ router, err := driver.NewRouter([]string{testDeviceID})
+ require.NoError(t, err)
+
client := &fakeClient{
inputs: make([]bool, 16),
coils: make([]bool, 32),
{Model: MixedModel},
},
}
- d, err := New(settings, publisher, client)
+ d, err := New(settings, publisher, router, client)
require.NoError(t, err)
- return d, client, transport
+ return d, client, transport, router
+}
+
+func testCommand(t *testing.T, capability string, value string) driver.Command {
+ t.Helper()
+
+ id, err := protocol.NewID(time.Now())
+ require.NoError(t, err)
+
+ expires := time.Now().UTC().Add(time.Minute)
+
+ return driver.Command{
+ DeviceID: testDeviceID,
+ Capability: capability,
+ Metadata: protocol.Metadata{
+ MessageID: id,
+ ExpiresAt: &expires,
+ },
+ Payload: protocol.ValuePayload{Value: []byte(value)},
+ }
}
func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
t.Helper()
- d, client, transport := testAssembly(t)
+ d, client, transport, _ := testAssembly(t)
connection := &connection{owner: d}
require.NoError(t, connection.start(t.Context()))
return connection, client, transport
}
+
+func queueCommands(t *testing.T, router *driver.Router, commands []driver.Command) {
+ t.Helper()
+
+ var (
+ publication protocol.Publication
+ err error
+ )
+
+ if len(commands) == 1 {
+ command := commands[0]
+ publication, err = protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+ } else {
+ members := make([]protocol.BatchCommand, len(commands))
+
+ for index, command := range commands {
+ members[index] = protocol.BatchCommand{
+ Capability: command.Capability,
+ Metadata: command.Metadata,
+ Payload: command.Payload,
+ }
+ }
+
+ publication, err = protocol.BuildCommandBatch(testDeviceID, commands[0].Metadata, protocol.BatchPayload{Commands: members})
+ }
+
+ require.NoError(t, err)
+ require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, false))
+}
--- /dev/null
+package m31
+
+import (
+ "bytes"
+ "encoding/json"
+ "math"
+ "math/big"
+ "time"
+
+ "onehouse/driver"
+)
+
+type outputRequest struct {
+ command driver.Command
+ point point
+ analog bool
+ value uint16
+}
+
+func newOutputRequest(command driver.Command, topology topology, deviceID string, batch bool) (outputRequest, string) {
+ request := outputRequest{command: command}
+
+ if command.DeviceID != deviceID {
+ return request, "Command belongs to another device"
+ }
+
+ if command.Expired(time.Now()) {
+ return request, "Command expired"
+ }
+
+ value := bytes.TrimSpace(command.Payload.Value)
+
+ for _, output := range topology.outputs {
+ if output.id != command.Capability {
+ continue
+ }
+
+ request.point = output
+ if !bytes.Equal(value, []byte("true")) && !bytes.Equal(value, []byte("false")) {
+ return request, "Digital output requires a Boolean value"
+ }
+
+ if bytes.Equal(value, []byte("true")) {
+ request.value = 1
+ }
+
+ if output.pulse != 0 && request.value != 1 {
+ return request, "Pulse output requires true"
+ }
+
+ if batch && output.pulse != 0 {
+ return request, "Batches require steady digital outputs"
+ }
+
+ return request, ""
+ }
+
+ for _, output := range topology.analog {
+ if output.id != command.Capability {
+ continue
+ }
+
+ request.point, request.analog = output, true
+ if batch {
+ return request, "Batches require steady digital outputs"
+ }
+
+ if len(value) == 0 || !json.Valid(value) || value[0] != '-' && (value[0] < '0' || value[0] > '9') {
+ return request, "Analog output requires zero through ten volts"
+ }
+
+ number, valid := new(big.Rat).SetString(string(value))
+ if !valid || number.Sign() < 0 || number.Cmp(big.NewRat(10, 1)) > 0 {
+ return request, "Analog output requires zero through ten volts"
+ }
+
+ voltage, _ := number.Float64()
+ request.value = uint16(math.Round(voltage * 1000))
+
+ return request, ""
+ }
+
+ return request, "Capability is not a configured output"
+}
--- /dev/null
+package m31
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestOutputRequestRejectsUnsafeValuesAndAccess(t *testing.T) {
+ d, _, _, _ := testAssembly(t)
+ testCases := []struct {
+ name string
+ point string
+ value string
+ batch bool
+ }{
+ {name: "rejects null Boolean", point: "m0.do0", value: "null"},
+ {name: "rejects numeric Boolean", point: "m0.do0", value: "1"},
+ {name: "rejects string Boolean", point: "m0.do0", value: `"true"`},
+ {name: "rejects input writes", point: "m0.di0", value: "true"},
+ {name: "rejects unknown points", point: "m8.do0", value: "true"},
+ {name: "rejects false pulses", point: "m0.do2", value: "false"},
+ {name: "rejects pulses in batches", point: "m0.do2", value: "true", batch: true},
+ {name: "rejects analog in batches", point: "m2.ao0", value: "5", batch: true},
+ {name: "rejects negative voltage", point: "m2.ao0", value: "-0.00000000000000001"},
+ {name: "rejects excessive voltage without float rounding", point: "m2.ao0", value: "10.00000000000000001"},
+ {name: "rejects string voltage", point: "m2.ao0", value: `"5"`},
+ {name: "rejects null voltage", point: "m2.ao0", value: "null"},
+ {name: "rejects malformed voltage", point: "m2.ao0", value: "5x"},
+ {name: "rejects missing voltage", point: "m2.ao0"},
+ {name: "rejects object voltage", point: "m2.ao0", value: "{}"},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ command := testCommand(t, tc.point, tc.value)
+ _, reason := newOutputRequest(command, d.topology, testDeviceID, tc.batch)
+ require.NotEmpty(t, reason)
+ })
+ }
+
+ command := testCommand(t, "m0.do0", "true")
+ command.DeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+ _, reason := newOutputRequest(command, d.topology, testDeviceID, false)
+ require.Contains(t, reason, "another device")
+
+ command.DeviceID = testDeviceID
+ expires := time.Now()
+ command.Metadata.ExpiresAt = &expires
+ _, reason = newOutputRequest(command, d.topology, testDeviceID, false)
+ require.Contains(t, reason, "expired")
+}
+
+func TestOutputRequestRoundsValidVoltageToMillivolts(t *testing.T) {
+ d, _, _, _ := testAssembly(t)
+ testCases := []struct {
+ name string
+ value string
+ want uint16
+ }{
+ {name: "accepts zero", value: "0"},
+ {name: "accepts full range", value: "10", want: 10000},
+ {name: "rounds half millivolt up", value: "5.5005", want: 5501},
+ {name: "accepts exponent notation", value: "55e-1", want: 5500},
+ {name: "accepts surrounding whitespace", value: " 5.5 ", want: 5500},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ request, reason := newOutputRequest(testCommand(t, "m2.ao0", tc.value), d.topology, testDeviceID, false)
+ require.Empty(t, reason)
+ require.Equal(t, tc.want, request.value)
+ })
+ }
+}
projects / onehouse / commitdiff