X-Git-Url: https://git.giacomolli.uk/onehouse/blobdiff_plain/f9e4f329fbb3381ad8f16ade26a62560e3c2b951..refs/heads/master:/driver/m31/driver.go?ds=sidebyside

diff --git a/driver/m31/driver.go b/driver/m31/driver.go
index 9201a59..8849c82 100644
--- a/driver/m31/driver.go
+++ b/driver/m31/driver.go
@@ -12,11 +12,13 @@ import (
 	"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
@@ -25,11 +27,14 @@ type Driver struct {
 	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")
 	}
@@ -57,6 +62,16 @@ func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver
 		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,
@@ -65,13 +80,15 @@ func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver
 		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 {
@@ -135,6 +152,7 @@ func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err err
 	}
 
 	healthy = true
+	handler := &commandHandler{connection: connection}
 	ticker := time.NewTicker(d.pollInterval)
 	defer ticker.Stop()
 
@@ -142,6 +160,11 @@ func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err err
 		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 {
@@ -155,10 +178,19 @@ 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
+	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)
+			}
+		}
 	}
 }