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{index: index, err: integration.Run(ctx)}
  35                 })
  36         }
  37
  38         defer workers.Wait()
  39
  40         var failures []error
  41
  42         for remaining := len(drivers); remaining > 0; {
  43                 select {
  44                 case _, open := <-connected:
  45                         if !open {
  46                                 connected = nil
  47
  48                                 continue
  49                         }
  50
  51                         for _, integration := range drivers {
  52                                 integration.Refresh()
  53                         }
  54                 case result := <-results:
  55                         remaining--
  56                         if result.err == nil || ctx.Err() != nil && stopOnly(result.err, ctx.Err()) {
  57                                 continue
  58                         }
  59
  60                         slog.Error("driver failed", "driver", result.index, "err", result.err)
  61
  62                         failures = append(failures, fmt.Errorf("run driver %d: %w", result.index, result.err))
  63
  64                         cancel()
  65                 }
  66         }
  67
  68         return errors.Join(failures...)
  69 }
  70
  71 type driverResult struct {
  72         index int
  73         err   error
  74 }
  75
  76 func stopOnly(err error, cancellation error) bool {
  77         joined, ok := err.(interface{ Unwrap() []error })
  78         if ok {
  79                 found := false
  80
  81                 for _, cause := range joined.Unwrap() {
  82                         if cause != nil {
  83                                 found = true
  84                                 if !stopOnly(cause, cancellation) {
  85                                         return false
  86                                 }
  87                         }
  88                 }
  89
  90                 return found
  91         }
  92
  93         wrapped, ok := err.(interface{ Unwrap() error })
  94         if ok {
  95                 return stopOnly(wrapped.Unwrap(), cancellation)
  96         }
  97
  98         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)
  99 }