Queue device commands and replay first results

The router parses each delivery once and keeps singles and ordered
batches in one bounded queue per device. Expiry, ownership, and caller
cancellation limit intake and queue backpressure. The device consumer
checks physical access and expiry again before work.

A shared per-device ledger retains the first 256 results with stable
metadata. Duplicate requests replay that result after uncertain delivery
without another action. Insertion order bounds memory, and callers
cannot mutate stored timestamps. The command and framework documents
state the signal behavior and the memory-only safety limits.

Issue: 40b314f

diff --git a/README.rst b/README.rst
index 84983f4d9610da4f21628c814dbd018c7d9c3bcb..41146534bf3c4e71b56283bf69df5c6aab8059b7 100644 (file)
--- a/README.rst
+++ b/README.rst
@@ -3,6 +3,9 @@
 ##########
 
 OneHouse is one program that controls a whole house. Linux is supported.
 ##########
 
 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.

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


@@ -27,3 +30,6 @@ Format all files, then run the checks:
    nix flake check
 
 ``AGENTS.md`` holds the rules for work in this repository.
    nix flake check
 
 ``AGENTS.md`` holds the rules for work in this repository.

+
+Run the command with ``nix run``. It waits for a signal without starting
+an integration or a broker.

diff --git a/doc/driver.rst b/doc/driver.rst
index 00b062ab63ce449a51636648e6d2a5e6a3d62cdd..6e13e21170be8e074e6b352927d7efe18808ab18 100644 (file)
--- a/doc/driver.rst
+++ b/doc/driver.rst
@@ -46,3 +46,34 @@ logged by the publisher. Reconnect refresh requests are nonblocking and
 coalesce while pending. The cache is memory-only. Tests use a fake
 transport and simulated time. They do not verify MQTT acknowledgement or
 a broker reconnect.
 coalesce while pending. The cache is memory-only. Tests use a fake
 transport and simulated time. They do not verify MQTT acknowledgement or
 a broker reconnect.

+
+*******************
+ Command ownership
+*******************
+
+``NewRouter`` takes the explicit IDs of command-capable devices. It
+supplies protocol subscriptions and one 64-entry FIFO per device. Each
+entry holds one single command or one ordered batch. Intake parses a
+delivery once through ``protocol.Parse``. A malformed or expired member
+rejects the whole batch. Retained commands and commands for unowned
+devices are errors. A full queue waits for space, cancellation, or the
+earliest member expiry. Each device has one queue consumer. It checks
+physical ownership, capability access, and expiry again before work.
+Read-only devices have no queue.
+
+``NewLedger`` creates a ledger for one device. The consumer checks
+``Lookup`` before a physical action, and stores its result with
+``Remember`` before the first publication attempt. A duplicate reuses
+the first status, reason, message ID, and observation time. It performs
+no second action. This also applies after a failed result send. Pass the
+returned ``ResultRecord`` fields to ``Publisher.Result`` unchanged.
+Records are safe from caller mutation.
+
+The ledger retains 256 results in insertion order per device. Lookup
+does not extend their age. It is not a reservation for concurrent
+physical actions. The single queue consumer provides that exclusion. The
+ledger is memory-only. Eviction or process restart removes duplicate
+protection for that record. Pending physical confirmation and
+device-specific command safety belong to the integration. Fake tests
+verify queue order, backpressure, expiry, per-device bounds, and stable
+duplicate replay without device I/O.

diff --git a/driver/command.go b/driver/command.go
new file mode 100644 (file)
index 0000000..49f7737
--- /dev/null
+++ b/driver/command.go
@@ -0,0 +1,22 @@
+package driver
+
+import (
+       "time"
+
+       "onehouse/protocol"
+)
+
+// Command is one validated device request. A driver checks ownership, access,
+// and Expired immediately before physical work. Each batch member keeps its own
+// metadata, expiry, and payload. The queue consumer owns these values.
+type Command struct {
+       DeviceID   string
+       Capability string
+       Metadata   protocol.Metadata
+       Payload    protocol.ValuePayload
+}
+
+// Expired reports whether the command has no expiry or has reached it.
+func (c Command) Expired(now time.Time) bool {
+       return c.Metadata.ExpiresAt == nil || !now.Before(*c.Metadata.ExpiresAt)
+}

diff --git a/driver/command_test.go b/driver/command_test.go
new file mode 100644 (file)
index 0000000..bf6998a
--- /dev/null
+++ b/driver/command_test.go
@@ -0,0 +1,19 @@
+package driver
+
+import (
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/protocol"
+)
+
+func TestCommandExpiryIncludesBoundaryAndMissingDeadline(t *testing.T) {
+       now := time.Now().UTC()
+       expiry := now.Add(time.Minute)
+       command := Command{Metadata: protocol.Metadata{ExpiresAt: &expiry}}
+       require.False(t, command.Expired(now))
+       require.True(t, command.Expired(expiry))
+       require.True(t, Command{}.Expired(now))
+}

diff --git a/driver/driver_result.go b/driver/driver_result.go
new file mode 100644 (file)
index 0000000..8ba9bbf
--- /dev/null
+++ b/driver/driver_result.go
@@ -0,0 +1,6 @@
+package driver
+
+type driverResult struct {
+       index int
+       err   error
+}

diff --git a/driver/ledger.go b/driver/ledger.go
new file mode 100644 (file)
index 0000000..8fb26af
--- /dev/null
+++ b/driver/ledger.go
@@ -0,0 +1,83 @@
+package driver
+
+import (
+       "fmt"
+       "sync"
+
+       "onehouse/protocol"
+)
+
+const commandResultCapacity = 256
+
+// Ledger keeps the first result of each of the most recent 256 commands for one
+// device, in insertion order. Its methods are safe for concurrent use. A single
+// queue consumer checks Lookup before acting and calls Remember before sending
+// the result, also after a failed send. The ledger does not reserve physical
+// work, persist records, or prevent replay after eviction.
+type Ledger struct {
+       deviceID string
+       mu       sync.Mutex
+       results  map[string]ResultRecord
+       order    []string
+}
+
+// NewLedger creates an empty result ledger for one canonical device ID.
+func NewLedger(deviceID string) (*Ledger, error) {
+       if !protocol.ValidID(deviceID) {
+               return nil, fmt.Errorf("ledger device ID must be a canonical ULID")
+       }
+
+       return &Ledger{
+               deviceID: deviceID,
+               results:  make(map[string]ResultRecord),
+               order:    make([]string, 0, commandResultCapacity),
+       }, nil
+}
+
+// Lookup returns an independent copy of a result without changing its age.
+func (l *Ledger) Lookup(commandID string) (ResultRecord, bool) {
+       l.mu.Lock()
+       defer l.mu.Unlock()
+
+       result, exists := l.results[commandID]
+
+       return result.clone(), exists
+}
+
+// Remember stores the first validated result before publication. An existing
+// result wins even when a duplicate proposes another status or reason. Its ID
+// and observation time remain stable after failed publication and on replay.
+func (l *Ledger) Remember(commandID string, payload protocol.ResultPayload) (ResultRecord, error) {
+       l.mu.Lock()
+       defer l.mu.Unlock()
+
+       result, exists := l.results[commandID]
+       if exists {
+               return result.clone(), nil
+       }
+
+       metadata, err := newMetadata(protocol.Metadata{CorrelationID: commandID})
+       if err != nil {
+               return ResultRecord{}, fmt.Errorf("create result metadata: %w", err)
+       }
+
+       _, err = protocol.BuildResult(l.deviceID, commandID, metadata, payload)
+       if err != nil {
+               return ResultRecord{}, fmt.Errorf("validate result: %w", err)
+       }
+
+       if len(l.order) == commandResultCapacity {
+               delete(l.results, l.order[0])
+
+               l.order = l.order[1:]
+       }
+
+       result = ResultRecord{
+               Metadata: metadata,
+               Payload:  payload,
+       }
+       l.results[commandID] = result
+       l.order = append(l.order, commandID)
+
+       return result.clone(), nil
+}

diff --git a/driver/ledger_test.go b/driver/ledger_test.go
new file mode 100644 (file)
index 0000000..28de6b2
--- /dev/null
+++ b/driver/ledger_test.go
@@ -0,0 +1,153 @@
+package driver
+
+import (
+       "errors"
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/protocol"
+)
+
+func TestLedgerReplaysFirstResultWithoutAnotherAction(t *testing.T) {
+       ledger, err := NewLedger(testDeviceID)
+       require.NoError(t, err)
+
+       transport := &recordingTransport{}
+       publisher := NewPublisher(transport)
+       actions := 0
+
+       var first ResultRecord
+
+       for attempt := range 3 {
+               result, exists := ledger.Lookup(testCommandID)
+               if !exists {
+                       actions++
+                       result, err = ledger.Remember(testCommandID, protocol.ResultPayload{Status: "completed"})
+                       require.NoError(t, err)
+
+                       first = result
+               }
+
+               if attempt == 0 {
+                       transport.setError(errors.New("delivery uncertain"))
+                       require.Error(t, publisher.Result(t.Context(), testDeviceID, testCommandID, result.Metadata, result.Payload))
+                       transport.setError(nil)
+               } else {
+                       require.NoError(t, publisher.Result(t.Context(), testDeviceID, testCommandID, result.Metadata, result.Payload))
+               }
+
+               require.Equal(t, first, result)
+       }
+
+       require.Equal(t, 1, actions)
+
+       packets := transport.messages()
+       require.Len(t, packets, 3)
+       require.Equal(t, packets[0], packets[1])
+       require.Equal(t, packets[0], packets[2])
+
+       ignored, err := ledger.Remember(testCommandID, protocol.ResultPayload{
+               Status: "failed",
+               Reason: "different outcome",
+       })
+       require.NoError(t, err)
+       require.Equal(t, first, ignored)
+}
+
+func TestLedgerEvictsOldestAndKeepsDevicesIndependent(t *testing.T) {
+       ledger, err := NewLedger(testDeviceID)
+       require.NoError(t, err)
+
+       first, err := ledger.Remember(testCommandID, protocol.ResultPayload{
+               Status: "rejected",
+               Reason: "invalid output",
+       })
+       require.NoError(t, err)
+
+       other, err := NewLedger(testOtherDeviceID)
+       require.NoError(t, err)
+
+       independent, err := other.Remember(testCommandID, protocol.ResultPayload{
+               Status: "failed",
+               Reason: "unavailable",
+       })
+       require.NoError(t, err)
+       require.NotEqual(t, first.Payload, independent.Payload)
+
+       lastID := ""
+
+       for range commandResultCapacity {
+               lastID, err = protocol.NewID(time.Now().UTC())
+               require.NoError(t, err)
+
+               _, err = ledger.Remember(lastID, protocol.ResultPayload{Status: "completed"})
+               require.NoError(t, err)
+       }
+
+       _, exists := ledger.Lookup(testCommandID)
+       require.False(t, exists)
+
+       _, exists = ledger.Lookup(lastID)
+       require.True(t, exists)
+
+       _, exists = other.Lookup(testCommandID)
+       require.True(t, exists)
+}
+
+func TestLedgerRejectsInvalidResultsAndCopiesReturnedTimes(t *testing.T) {
+       _, err := NewLedger("invalid")
+       require.Error(t, err)
+
+       ledger, err := NewLedger(testDeviceID)
+       require.NoError(t, err)
+
+       testCases := []struct {
+               name    string
+               payload protocol.ResultPayload
+       }{
+               {
+                       name:    "failure requires a reason",
+                       payload: protocol.ResultPayload{Status: "failed"},
+               },
+               {
+                       name: "completion rejects a reason",
+                       payload: protocol.ResultPayload{
+                               Status: "completed",
+                               Reason: "not allowed",
+                       },
+               },
+               {
+                       name:    "unknown status rejects",
+                       payload: protocol.ResultPayload{Status: "invalid"},
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       _, err = ledger.Remember(testCommandID, tc.payload)
+                       require.Error(t, err)
+
+                       _, exists := ledger.Lookup(testCommandID)
+                       require.False(t, exists)
+               })
+       }
+
+       _, err = ledger.Remember("invalid", protocol.ResultPayload{Status: "completed"})
+       require.Error(t, err)
+
+       result, err := ledger.Remember(testCommandID, protocol.ResultPayload{Status: "completed"})
+       require.NoError(t, err)
+
+       observation := *result.Metadata.SentAt
+       *result.Metadata.SentAt = observation.Add(time.Hour)
+       first, exists := ledger.Lookup(testCommandID)
+       require.True(t, exists)
+       require.Equal(t, observation, *first.Metadata.SentAt)
+
+       *first.Metadata.SentAt = observation.Add(time.Hour)
+       second, err := ledger.Remember(testCommandID, protocol.ResultPayload{})
+       require.NoError(t, err)
+       require.Equal(t, observation, *second.Metadata.SentAt)
+}

diff --git a/driver/publisher_test.go b/driver/publisher_test.go
index d1135b5ac25b92d4ad51c89d8b8391647dd4fc41..642f6ab8e526bdc20a2c8f14dc8045fc7a26c8f7 100644 (file)
--- a/driver/publisher_test.go
+++ b/driver/publisher_test.go
@@ -342,6 +342,35 @@ func TestPublisherPreservesDirectMeasurementEvidenceAtRepeatDeadline(t *testing.
        }
 }
 
        }
 }
 

+func TestPublisherRunsAsDriverAndRefreshesOnReconnect(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               transport := &recordingTransport{}
+               publisher := NewPublisher(transport)
+               require.NoError(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+
+               connected := make(chan struct{}, 1)
+               finished := make(chan error, 1)
+               go func() {
+                       finished <- Run(ctx, []Driver{publisher}, connected)
+               }()
+
+               synctest.Wait()
+
+               connected <- struct{}{}
+
+               synctest.Wait()
+
+               packets := transport.messages()
+               require.Len(t, packets, 2)
+               require.NotEqual(t, packets[0].Data, packets[1].Data)
+               cancel()
+               require.NoError(t, <-finished)
+       })
+}
+

 type waitingTransport struct{}
 
 func (p *waitingTransport) Publish(ctx context.Context, _ protocol.Publication) error {
 type waitingTransport struct{}
 
 func (p *waitingTransport) Publish(ctx context.Context, _ protocol.Publication) error {

diff --git a/driver/result_record.go b/driver/result_record.go
new file mode 100644 (file)
index 0000000..65a60e2
--- /dev/null
+++ b/driver/result_record.go
@@ -0,0 +1,20 @@
+package driver
+
+import "onehouse/protocol"
+
+// ResultRecord holds the first result and its stable replay metadata. Pass both
+// fields unchanged to Publisher.Result. Lookup and Remember return independent
+// timestamp pointers, so callers cannot change the ledger.
+type ResultRecord struct {
+       Metadata protocol.Metadata
+       Payload  protocol.ResultPayload
+}
+
+func (r ResultRecord) clone() ResultRecord {
+       if r.Metadata.SentAt != nil {
+               timestamp := *r.Metadata.SentAt
+               r.Metadata.SentAt = &timestamp
+       }
+
+       return r
+}

diff --git a/driver/router.go b/driver/router.go
new file mode 100644 (file)
index 0000000..052886a
--- /dev/null
+++ b/driver/router.go
@@ -0,0 +1,135 @@
+package driver
+
+import (
+       "context"
+       "fmt"
+       "slices"
+       "time"
+
+       "onehouse/protocol"
+)
+
+const commandQueueCapacity = 64
+
+// Router owns one bounded FIFO for each supplied command-capable device. A
+// queue entry is one single command or one ordered batch. Handle is safe for
+// concurrent calls. Each queue has one consumer that serializes device work.
+// The caller stops producers and consumers through their contexts. Queues do
+// not close and belong to the router for its lifetime.
+type Router struct {
+       queues        map[string]chan []Command
+       subscriptions []protocol.Subscription
+}
+
+// NewRouter validates unique device IDs and makes 64-entry queues. Read-only
+// devices do not belong in deviceIDs. The constructor performs no I/O.
+func NewRouter(deviceIDs []string) (*Router, error) {
+       ids := slices.Clone(deviceIDs)
+       slices.Sort(ids)
+
+       router := &Router{queues: make(map[string]chan []Command, len(deviceIDs))}
+
+       for _, deviceID := range ids {
+               _, exists := router.queues[deviceID]
+               if exists {
+                       return nil, fmt.Errorf("duplicate command device ID")
+               }
+
+               subscription, err := protocol.CommandSubscription(deviceID)
+               if err != nil {
+                       return nil, fmt.Errorf("build command subscription: %w", err)
+               }
+
+               router.queues[deviceID] = make(chan []Command, commandQueueCapacity)
+               router.subscriptions = append(router.subscriptions, subscription)
+       }
+
+       return router, nil
+}
+
+// Subscriptions returns an independent, sorted list including batch intake.
+func (r *Router) Subscriptions() []protocol.Subscription {
+       return slices.Clone(r.subscriptions)
+}
+
+// Queue returns the receive-only command queue for a configured device.
+func (r *Router) Queue(deviceID string) (<-chan []Command, bool) {
+       queue, exists := r.queues[deviceID]
+
+       return queue, exists
+}
+
+// Handle parses one delivery once and queues all its commands together. It
+// rejects malformed, retained, expired, non-command, and unowned deliveries. A
+// full queue waits until space, caller cancellation, or the earliest member
+// expiry. A driver must check expiry again when it consumes the queue entry.
+func (r *Router) Handle(ctx context.Context, topic string, data []byte, retained bool) error {
+       err := ctx.Err()
+       if err != nil {
+               return err
+       }
+
+       message, err := protocol.Parse(topic, data, retained)
+       if err != nil {
+               return fmt.Errorf("parse command delivery: %w", err)
+       }
+
+       var commands []Command
+
+       switch payload := message.Payload.(type) {
+       case protocol.ValuePayload:
+               if message.Address.Kind != protocol.Command {
+                       return fmt.Errorf("delivery is not a command")
+               }
+
+               commands = []Command{{
+                       DeviceID:   message.Address.ID,
+                       Capability: message.Address.Target,
+                       Metadata:   message.Metadata,
+                       Payload:    payload,
+               }}
+       case protocol.BatchPayload:
+               for _, member := range payload.Commands {
+                       commands = append(commands, Command{
+                               DeviceID:   message.Address.ID,
+                               Capability: member.Capability,
+                               Metadata:   member.Metadata,
+                               Payload:    member.Payload,
+                       })
+               }
+       default:
+               return fmt.Errorf("delivery is not a command")
+       }
+
+       queue, exists := r.queues[message.Address.ID]
+       if !exists {
+               return fmt.Errorf("command device has no queue")
+       }
+
+       expiresAt := *commands[0].Metadata.ExpiresAt
+
+       for _, command := range commands {
+               if command.Expired(time.Now()) {
+                       return fmt.Errorf("command expired")
+               }
+
+               if command.Metadata.ExpiresAt.Before(expiresAt) {
+                       expiresAt = *command.Metadata.ExpiresAt
+               }
+       }
+
+       ctx, cancel := context.WithDeadline(ctx, expiresAt)
+       defer cancel()
+
+       err = ctx.Err()
+       if err != nil {
+               return err
+       }
+
+       select {
+       case queue <- commands:
+               return nil
+       case <-ctx.Done():
+               return ctx.Err()
+       }
+}

diff --git a/driver/router_test.go b/driver/router_test.go
new file mode 100644 (file)
index 0000000..5e9a3e3
--- /dev/null
+++ b/driver/router_test.go
@@ -0,0 +1,272 @@
+package driver
+
+import (
+       "context"
+       "encoding/json"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/protocol"
+)
+
+func TestRouterKeepsSinglesAndBatchesInOneDeviceOrder(t *testing.T) {
+       router, err := NewRouter([]string{testDeviceID})
+       require.NoError(t, err)
+
+       queue, exists := router.Queue(testDeviceID)
+       require.True(t, exists)
+
+       first := routerCommand(t, testDeviceID, "relay", time.Minute)
+       second := routerCommand(t, testDeviceID, "output", time.Minute)
+       third := routerCommand(t, testDeviceID, "pulse", time.Minute)
+       last := routerCommand(t, testDeviceID, "batch", time.Minute)
+       batch, err := protocol.BuildCommandBatch(testDeviceID, second.Metadata, protocol.BatchPayload{
+               Commands: []protocol.BatchCommand{
+                       {
+                               Metadata:   second.Metadata,
+                               Capability: second.Capability,
+                               Payload:    second.Payload,
+                       },
+                       {
+                               Metadata:   third.Metadata,
+                               Capability: third.Capability,
+                               Payload:    third.Payload,
+                       },
+               },
+       })
+       require.NoError(t, err)
+
+       for _, publication := range []protocol.Publication{commandPublication(t, first), batch, commandPublication(t, last)} {
+               require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, publication.Retain))
+       }
+
+       require.Len(t, queue, 3)
+       require.Equal(t, []Command{first}, <-queue)
+       require.Equal(t, []Command{second, third}, <-queue)
+       require.Equal(t, []Command{last}, <-queue)
+}
+
+func TestRouterBuildsOnlyExplicitDeviceSubscriptions(t *testing.T) {
+       router, err := NewRouter([]string{testOtherDeviceID, testDeviceID})
+       require.NoError(t, err)
+
+       first, err := protocol.CommandSubscription(testDeviceID)
+       require.NoError(t, err)
+
+       second, err := protocol.CommandSubscription(testOtherDeviceID)
+       require.NoError(t, err)
+
+       subscriptions := router.Subscriptions()
+       require.Equal(t, []protocol.Subscription{first, second}, subscriptions)
+
+       subscriptions[0].Filter = "changed"
+
+       require.Equal(t, first, router.Subscriptions()[0])
+
+       queue, exists := router.Queue("unknown")
+       require.False(t, exists)
+       require.Nil(t, queue)
+
+       _, err = NewRouter([]string{"invalid"})
+       require.Error(t, err)
+
+       _, err = NewRouter([]string{testDeviceID, testDeviceID})
+       require.Error(t, err)
+
+       readOnly, err := NewRouter(nil)
+       require.NoError(t, err)
+       require.Empty(t, readOnly.Subscriptions())
+}
+
+func TestRouterRejectsInvalidUnownedAndExpiredDeliveries(t *testing.T) {
+       router, err := NewRouter([]string{testDeviceID})
+       require.NoError(t, err)
+
+       queue, exists := router.Queue(testDeviceID)
+       require.True(t, exists)
+
+       valid := commandPublication(t, routerCommand(t, testDeviceID, "relay", time.Minute))
+       unowned := commandPublication(t, routerCommand(t, testOtherDeviceID, "relay", time.Minute))
+       expired := commandPublication(t, routerCommand(t, testDeviceID, "relay", -time.Second))
+       value, err := protocol.BuildReported(testDeviceID, "relay", protocolMetadata(t), protocol.ValuePayload{Value: json.RawMessage("true")})
+       require.NoError(t, err)
+
+       availability, err := protocol.BuildAvailability(testDeviceID, protocolMetadata(t), protocol.AvailabilityPayload{Status: "online"})
+       require.NoError(t, err)
+
+       testCases := []struct {
+               name        string
+               publication protocol.Publication
+       }{
+               {
+                       name: "retained command rejects",
+                       publication: protocol.Publication{
+                               Topic:  valid.Topic,
+                               Data:   valid.Data,
+                               Retain: true,
+                       },
+               },
+               {
+                       name: "malformed payload rejects",
+                       publication: protocol.Publication{
+                               Topic: valid.Topic,
+                               Data:  []byte(`{}`),
+                       },
+               },
+               {
+                       name: "unknown topic rejects",
+                       publication: protocol.Publication{
+                               Topic: "invalid",
+                               Data:  valid.Data,
+                       },
+               },
+               {
+                       name:        "unowned device rejects",
+                       publication: unowned,
+               },
+               {
+                       name:        "expired command rejects",
+                       publication: expired,
+               },
+               {
+                       name:        "reported value is not command intake",
+                       publication: value,
+               },
+               {
+                       name:        "availability is not command intake",
+                       publication: availability,
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       require.Error(t, router.Handle(t.Context(), tc.publication.Topic, tc.publication.Data, tc.publication.Retain))
+                       require.Empty(t, queue)
+               })
+       }
+}
+
+func TestRouterRejectsWholeBatchWhenOneMemberExpires(t *testing.T) {
+       router, err := NewRouter([]string{testDeviceID})
+       require.NoError(t, err)
+
+       first := routerCommand(t, testDeviceID, "relay", time.Minute)
+       second := routerCommand(t, testDeviceID, "pulse", -time.Second)
+       batch, err := protocol.BuildCommandBatch(testDeviceID, first.Metadata, protocol.BatchPayload{
+               Commands: []protocol.BatchCommand{
+                       {
+                               Metadata:   first.Metadata,
+                               Capability: first.Capability,
+                               Payload:    first.Payload,
+                       },
+                       {
+                               Metadata:   second.Metadata,
+                               Capability: second.Capability,
+                               Payload:    second.Payload,
+                       },
+               },
+       })
+       require.NoError(t, err)
+       require.Error(t, router.Handle(t.Context(), batch.Topic, batch.Data, false))
+
+       queue, exists := router.Queue(testDeviceID)
+       require.True(t, exists)
+       require.Empty(t, queue)
+}
+
+func TestRouterBackpressureHonorsCancellationAndEarliestExpiry(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               router, err := NewRouter([]string{testDeviceID})
+               require.NoError(t, err)
+
+               valid := commandPublication(t, routerCommand(t, testDeviceID, "relay", time.Hour))
+
+               for range commandQueueCapacity {
+                       require.NoError(t, router.Handle(t.Context(), valid.Topic, valid.Data, false))
+               }
+
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               finished := make(chan error, 1)
+               go func() {
+                       finished <- router.Handle(ctx, valid.Topic, valid.Data, false)
+               }()
+
+               synctest.Wait()
+               require.Empty(t, finished)
+               cancel()
+               require.ErrorIs(t, <-finished, context.Canceled)
+
+               short := commandPublication(t, routerCommand(t, testDeviceID, "pulse", time.Second))
+               require.ErrorIs(t, router.Handle(t.Context(), short.Topic, short.Data, false), context.DeadlineExceeded)
+
+               first := routerCommand(t, testDeviceID, "relay", time.Minute)
+               second := routerCommand(t, testDeviceID, "pulse", time.Second)
+               batch, err := protocol.BuildCommandBatch(testDeviceID, first.Metadata, protocol.BatchPayload{Commands: []protocol.BatchCommand{
+                       {
+                               Metadata:   first.Metadata,
+                               Capability: first.Capability,
+                               Payload:    first.Payload,
+                       },
+                       {
+                               Metadata:   second.Metadata,
+                               Capability: second.Capability,
+                               Payload:    second.Payload,
+                       },
+               }})
+               require.NoError(t, err)
+
+               start := time.Now()
+
+               require.ErrorIs(t, router.Handle(t.Context(), batch.Topic, batch.Data, false), context.DeadlineExceeded)
+               require.Equal(t, time.Second, time.Since(start))
+
+               queue, exists := router.Queue(testDeviceID)
+               require.True(t, exists)
+               require.Len(t, queue, commandQueueCapacity)
+               require.ErrorIs(t, router.Handle(ctx, valid.Topic, valid.Data, false), context.Canceled)
+               <-queue
+               require.NoError(t, router.Handle(t.Context(), valid.Topic, valid.Data, false))
+       })
+}
+
+func routerCommand(t *testing.T, deviceID string, capability string, lifetime time.Duration) Command {
+       t.Helper()
+
+       metadata := protocolMetadata(t)
+       expiry := time.Now().UTC().Add(lifetime)
+       metadata.ExpiresAt = &expiry
+
+       return Command{
+               DeviceID:   deviceID,
+               Capability: capability,
+               Metadata:   metadata,
+               Payload:    protocol.ValuePayload{Value: json.RawMessage("true")},
+       }
+}
+
+func commandPublication(t *testing.T, command Command) protocol.Publication {
+       t.Helper()
+
+       publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+       require.NoError(t, err)
+
+       return publication
+}
+
+func protocolMetadata(t *testing.T) protocol.Metadata {
+       t.Helper()
+
+       now := time.Now().UTC()
+       id, err := protocol.NewID(now)
+       require.NoError(t, err)
+
+       return protocol.Metadata{
+               MessageID: id,
+               SentAt:    &now,
+       }
+}

diff --git a/driver/runner.go b/driver/runner.go
index 287c333cb75cb7706cf6a82720bc23fd24ab5dc9..b40702bb0267331083e57789fbecf4cf4e13b8ab 100644 (file)
--- a/driver/runner.go
+++ b/driver/runner.go
@@ -71,11 +71,6 @@ func Run(ctx context.Context, drivers []Driver, connected <-chan struct{}) error
        return errors.Join(failures...)
 }
 
        return errors.Join(failures...)
 }
 

-type driverResult struct {
-       index int
-       err   error
-}
-

 func stopOnly(err error, cancellation error) bool {
        joined, ok := err.(interface{ Unwrap() []error })
        if ok {
 func stopOnly(err error, cancellation error) bool {
        joined, ok := err.(interface{ Unwrap() []error })
        if ok {