git.giacomolli.uk - onehouse/blob - driver/m31/driver.go

   1 package m31
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "net"
   8         "strconv"
   9         "time"
  10
  11         "onehouse/driver"
  12         "onehouse/protocol"
  13 )
  14
  15 // Driver serializes one assembly's polling and commands. Run has one owner and
  16 // closes the supplied Client on each attempt. Recoverable device failures
  17 // publish offline and retry with backoff. Publication or cleanup failures
  18 // return to the framework runner. The shared publisher must run alongside this
  19 // driver. Refresh is safe during Run. Duplicate protection ends at ledger
  20 // eviction or process restart. No operation starts a physical connection before
  21 // Run.
  22 type Driver struct {
  23         deviceID     string
  24         address      string
  25         unit         byte
  26         pollInterval time.Duration
  27         topology     topology
  28         publisher    *driver.Publisher
  29         client       Client
  30         queue        <-chan []driver.Command
  31         ledger       *driver.Ledger
  32 }
  33
  34 // New validates settings and requires the device's queue in router. The caller
  35 // supplies valid publisher, router, and client dependencies and starts the
  36 // router's intake separately. The driver owns the client until Run returns.
  37 func New(settings Settings, publisher *driver.Publisher, router *driver.Router, client Client) (*Driver, error) {
  38         if !protocol.ValidID(settings.DeviceID) {
  39                 return nil, fmt.Errorf("device ID must be a canonical ULID")
  40         }
  41
  42         host, port, err := net.SplitHostPort(settings.Address)
  43         if err != nil || host == "" {
  44                 return nil, fmt.Errorf("address must contain a host and TCP port")
  45         }
  46
  47         number, err := strconv.Atoi(port)
  48         if err != nil || number < 1 || number > 65535 {
  49                 return nil, fmt.Errorf("TCP port is outside its range")
  50         }
  51
  52         if settings.Unit == 0 || settings.Unit > 247 {
  53                 return nil, fmt.Errorf("unit must be from one through 247")
  54         }
  55
  56         if settings.PollInterval < 10*time.Millisecond || settings.PollInterval > time.Minute {
  57                 return nil, fmt.Errorf("poll interval must be from ten milliseconds through one minute")
  58         }
  59
  60         points, err := newTopology(settings.Modules)
  61         if err != nil {
  62                 return nil, fmt.Errorf("validate topology: %w", err)
  63         }
  64
  65         queue, exists := router.Queue(settings.DeviceID)
  66         if !exists {
  67                 return nil, fmt.Errorf("device has no command queue")
  68         }
  69
  70         ledger, err := driver.NewLedger(settings.DeviceID)
  71         if err != nil {
  72                 return nil, fmt.Errorf("create result ledger: %w", err)
  73         }
  74
  75         return &Driver{
  76                 deviceID:     settings.DeviceID,
  77                 address:      settings.Address,
  78                 unit:         settings.Unit,
  79                 pollInterval: settings.PollInterval,
  80                 topology:     points,
  81                 publisher:    publisher,
  82                 client:       client,
  83                 queue:        queue,
  84                 ledger:       ledger,
  85         }, nil
  86 }
  87
  88 // Run configures and verifies output modes before online or command work. A
  89 // successful first read establishes a baseline without input events. Device
  90 // failures retry after one second, doubling to thirty seconds until a complete
  91 // baseline succeeds. Commands received during backoff fail without I/O.
  92 func (d *Driver) Run(ctx context.Context) error {
  93         err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
  94         if err != nil {
  95                 return fmt.Errorf("publish descriptor: %w", err)
  96         }
  97
  98         delay := time.Second
  99
 100         for {
 101                 healthy, retry, err := d.attempt(ctx)
 102                 if !retry || ctx.Err() != nil {
 103                         return fmt.Errorf("run assembly: %w", err)
 104                 }
 105
 106                 err = d.publisher.Availability(ctx, d.deviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"})
 107                 if err != nil {
 108                         return fmt.Errorf("publish offline: %w", err)
 109                 }
 110
 111                 if healthy {
 112                         delay = time.Second
 113                 }
 114
 115                 err = d.wait(ctx, delay)
 116                 if err != nil {
 117                         return fmt.Errorf("wait for reconnect: %w", err)
 118                 }
 119
 120                 delay = min(delay*2, 30*time.Second)
 121         }
 122 }
 123
 124 // Refresh requests retained snapshots through the shared publisher. It does not
 125 // repeat input events, physical commands, or pulse operations.
 126 func (d *Driver) Refresh() {
 127         d.publisher.Refresh()
 128 }
 129
 130 func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err error) {
 131         defer func() {
 132                 var failure *deviceError
 133
 134                 retry = errors.As(err, &failure)
 135
 136                 closeErr := d.client.Close()
 137                 if closeErr != nil {
 138                         retry = false
 139                         err = fmt.Errorf("close assembly: %w", errors.Join(err, closeErr))
 140                 }
 141         }()
 142
 143         err = d.client.Connect(ctx, d.address, d.unit)
 144         if err != nil {
 145                 return false, false, &deviceError{err: fmt.Errorf("connect assembly: %w", err)}
 146         }
 147
 148         connection := &connection{owner: d}
 149         err = connection.start(ctx)
 150         if err != nil {
 151                 return false, false, fmt.Errorf("initialize assembly: %w", err)
 152         }
 153
 154         healthy = true
 155         handler := &commandHandler{connection: connection}
 156         ticker := time.NewTicker(d.pollInterval)
 157         defer ticker.Stop()
 158
 159         for {
 160                 select {
 161                 case <-ctx.Done():
 162                         return healthy, false, ctx.Err()
 163                 case commands := <-d.queue:
 164                         err = handler.handle(ctx, commands)
 165                         if err != nil {
 166                                 return healthy, false, fmt.Errorf("process commands: %w", err)
 167                         }
 168                 case <-ticker.C:
 169                         err = connection.poll(ctx)
 170                         if err != nil {
 171                                 return healthy, false, fmt.Errorf("poll assembly: %w", err)
 172                         }
 173                 }
 174         }
 175 }
 176
 177 func (d *Driver) wait(ctx context.Context, delay time.Duration) error {
 178         timer := time.NewTimer(delay)
 179         defer timer.Stop()
 180
 181         handler := &commandHandler{connection: &connection{owner: d}}
 182
 183         for {
 184                 select {
 185                 case <-ctx.Done():
 186                         return ctx.Err()
 187                 case <-timer.C:
 188                         return nil
 189                 case commands := <-d.queue:
 190                         err := handler.unavailable(ctx, commands)
 191                         if err != nil {
 192                                 return fmt.Errorf("finish unavailable commands: %w", err)
 193                         }
 194                 }
 195         }
 196 }