Share the Alamo gateway request budget

[onehouse] / driver / m31 / driver.go

diff --git a/driver/m31/driver.go b/driver/m31/driver.go
index 9201a59d6e9d7b5e9db5d17bd14f5e96092b3bbd..8849c824837613e50b0c27a4ffac20f2f3b7f873 100644 (file)
--- a/driver/m31/driver.go
+++ b/driver/m31/driver.go
@@ -12,11 +12,13 @@ import (
        "onehouse/protocol"
 )
 
        "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
 type Driver struct {
        deviceID     string
        address      string


@@ -25,11 +27,14 @@ type Driver struct {
        topology     topology
        publisher    *driver.Publisher
        client       Client
        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")
        }
        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)
        }
 
                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,
        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,
                topology:     points,
                publisher:    publisher,
                client:       client,

+               queue:        queue,
+               ledger:       ledger,

        }, nil
 }
 
        }, 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
 // 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 {
 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
        }
 
        healthy = true

+       handler := &commandHandler{connection: connection}

        ticker := time.NewTicker(d.pollInterval)
        defer ticker.Stop()
 
        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()
                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 {
                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()
 
        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)
+                       }
+               }

        }
 }
        }
 }