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

   1 package driver
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "io"
   8         "log/slog"
   9         "net"
  10         "sync"
  11         "syscall"
  12 )
  13
  14 // Run starts the explicit driver list and waits for every driver to stop. A
  15 // failure cancels the other drivers. Reconnect notifications request fresh
  16 // snapshots. A nil or closed connected channel disables notifications. An empty
  17 // list waits for cancellation. Only this runner logs driver failures.
  18 func Run(ctx context.Context, drivers []Driver, connected <-chan struct{}) error {
  19         ctx, cancel := context.WithCancel(ctx)
  20         defer cancel()
  21
  22         if len(drivers) == 0 {
  23                 <-ctx.Done()
  24
  25                 return nil
  26         }
  27
  28         results := make(chan driverResult, len(drivers))
  29
  30         var workers sync.WaitGroup
  31
  32         for index, integration := range drivers {
  33                 workers.Go(func() {
  34                         results <- driverResult{
  35                                 index: index,
  36                                 err:   integration.Run(ctx),
  37                         }
  38                 })
  39         }
  40
  41         defer workers.Wait()
  42
  43         var failures []error
  44
  45         for remaining := len(drivers); remaining > 0; {
  46                 select {
  47                 case _, open := <-connected:
  48                         if !open {
  49                                 connected = nil
  50
  51                                 continue
  52                         }
  53
  54                         for _, integration := range drivers {
  55                                 integration.Refresh()
  56                         }
  57                 case result := <-results:
  58                         remaining--
  59                         if result.err == nil || ctx.Err() != nil && stopOnly(result.err, ctx.Err()) {
  60                                 continue
  61                         }
  62
  63                         slog.Error("driver failed", "driver", result.index, "err", result.err)
  64
  65                         failures = append(failures, fmt.Errorf("run driver %d: %w", result.index, result.err))
  66
  67                         cancel()
  68                 }
  69         }
  70
  71         return errors.Join(failures...)
  72 }
  73
  74 func stopOnly(err error, cancellation error) bool {
  75         joined, ok := err.(interface{ Unwrap() []error })
  76         if ok {
  77                 found := false
  78
  79                 for _, cause := range joined.Unwrap() {
  80                         if cause != nil {
  81                                 found = true
  82                                 if !stopOnly(cause, cancellation) {
  83                                         return false
  84                                 }
  85                         }
  86                 }
  87
  88                 return found
  89         }
  90
  91         wrapped, ok := err.(interface{ Unwrap() error })
  92         if ok {
  93                 return stopOnly(wrapped.Unwrap(), cancellation)
  94         }
  95
  96         return errors.Is(err, cancellation) || errors.Is(err, net.ErrClosed) || errors.Is(err, io.ErrClosedPipe) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET)
  97 }