--- /dev/null
+package driver
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net"
+ "sync"
+ "syscall"
+)
+
+// Run starts the explicit driver list and waits for every driver to stop. A
+// failure cancels the other drivers. Reconnect notifications request fresh
+// snapshots. A nil or closed connected channel disables notifications. An empty
+// list waits for cancellation. Only this runner logs driver failures.
+func Run(ctx context.Context, drivers []Driver, connected <-chan struct{}) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ if len(drivers) == 0 {
+ <-ctx.Done()
+
+ return nil
+ }
+
+ results := make(chan driverResult, len(drivers))
+
+ var workers sync.WaitGroup
+
+ for index, integration := range drivers {
+ workers.Go(func() {
+ results <- driverResult{index: index, err: integration.Run(ctx)}
+ })
+ }
+
+ defer workers.Wait()
+
+ var failures []error
+
+ for remaining := len(drivers); remaining > 0; {
+ select {
+ case _, open := <-connected:
+ if !open {
+ connected = nil
+
+ continue
+ }
+
+ for _, integration := range drivers {
+ integration.Refresh()
+ }
+ case result := <-results:
+ remaining--
+ if result.err == nil || ctx.Err() != nil && stopOnly(result.err, ctx.Err()) {
+ continue
+ }
+
+ slog.Error("driver failed", "driver", result.index, "err", result.err)
+
+ failures = append(failures, fmt.Errorf("run driver %d: %w", result.index, result.err))
+
+ cancel()
+ }
+ }
+
+ return errors.Join(failures...)
+}
+
+type driverResult struct {
+ index int
+ err error
+}
+
+func stopOnly(err error, cancellation error) bool {
+ joined, ok := err.(interface{ Unwrap() []error })
+ if ok {
+ found := false
+
+ for _, cause := range joined.Unwrap() {
+ if cause != nil {
+ found = true
+ if !stopOnly(cause, cancellation) {
+ return false
+ }
+ }
+ }
+
+ return found
+ }
+
+ wrapped, ok := err.(interface{ Unwrap() error })
+ if ok {
+ return stopOnly(wrapped.Unwrap(), cancellation)
+ }
+
+ 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)
+}
--- /dev/null
+package driver
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net"
+ "syscall"
+ "testing"
+ "testing/synctest"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestRunRefreshesDriversAndWaitsForCleanup(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ first := &fakeDriver{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
+ second := &fakeDriver{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
+ connected := make(chan struct{}, 1)
+ finished := make(chan error, 1)
+ go func() {
+ finished <- Run(ctx, []Driver{first, second}, connected)
+ }()
+
+ synctest.Wait()
+
+ connected <- struct{}{}
+
+ synctest.Wait()
+ require.Len(t, first.refreshed, 1)
+ require.Len(t, second.refreshed, 1)
+ close(connected)
+ synctest.Wait()
+ require.Empty(t, finished)
+ cancel()
+ require.NoError(t, <-finished)
+ <-first.stopped
+ <-second.stopped
+ })
+}
+
+func TestRunClassifiesStopsAndJoinedFailures(t *testing.T) {
+ failure := errors.New("readback failed")
+ testCases := []struct {
+ name string
+ err error
+ canceled bool
+ failure bool
+ }{
+ {name: "normal completion", err: nil},
+ {name: "cancellation is a stop", err: context.Canceled, canceled: true},
+ {name: "closed transport is a stop", err: net.ErrClosed, canceled: true},
+ {name: "closed pipe is a stop", err: io.ErrClosedPipe, canceled: true},
+ {name: "broken pipe is a stop", err: syscall.EPIPE, canceled: true},
+ {name: "reset connection is a stop", err: syscall.ECONNRESET, canceled: true},
+ {name: "wrapped stop stays a stop", err: fmt.Errorf("read: %w", net.ErrClosed), canceled: true},
+ {name: "joined stops stay stops", err: errors.Join(context.Canceled, net.ErrClosed), canceled: true},
+ {name: "joined failure stays a failure", err: errors.Join(context.Canceled, failure), canceled: true, failure: true},
+ {name: "request timeout stays a failure", err: context.DeadlineExceeded, failure: true},
+ {name: "active transport close is a failure", err: net.ErrClosed, failure: true},
+ {name: "independent failure stays a failure", err: failure, canceled: true, failure: true},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ var output bytes.Buffer
+
+ previous := slog.Default()
+
+ slog.SetDefault(slog.New(slog.NewTextHandler(&output, nil)))
+ t.Cleanup(func() {
+ slog.SetDefault(previous)
+ })
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ if tc.canceled {
+ cancel()
+ }
+
+ integration := &resultDriver{err: tc.err}
+ err := Run(ctx, []Driver{integration}, nil)
+ if tc.failure {
+ require.ErrorIs(t, err, tc.err)
+ require.Contains(t, output.String(), "level=ERROR")
+ } else {
+ require.NoError(t, err)
+ require.Empty(t, output.String())
+ }
+ })
+ }
+}
+
+func TestRunCancelsPeersAndKeepsTheirIndependentFailure(t *testing.T) {
+ firstFailure := errors.New("first failed")
+ cleanupFailure := errors.New("cleanup failed")
+ first := &resultDriver{err: firstFailure}
+ second := &fakeDriver{
+ stopped: make(chan struct{}),
+ refreshed: make(chan struct{}, 1),
+ err: cleanupFailure,
+ }
+ err := Run(t.Context(), []Driver{first, second}, nil)
+ require.ErrorIs(t, err, firstFailure)
+ require.ErrorIs(t, err, cleanupFailure)
+ <-second.stopped
+}
+
+func TestRunEmptyListWaitsForCancellation(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ finished := make(chan error, 1)
+ go func() {
+ finished <- Run(ctx, nil, nil)
+ }()
+
+ synctest.Wait()
+ require.Empty(t, finished)
+ cancel()
+ require.NoError(t, <-finished)
+ })
+}
+
+type fakeDriver struct {
+ stopped chan struct{}
+ refreshed chan struct{}
+ err error
+}
+
+func (d *fakeDriver) Run(ctx context.Context) error {
+ <-ctx.Done()
+
+ defer close(d.stopped)
+
+ return errors.Join(ctx.Err(), d.err)
+}
+
+func (d *fakeDriver) Refresh() {
+ d.refreshed <- struct{}{}
+}
+
+type resultDriver struct {
+ err error
+}
+
+func (d *resultDriver) Run(_ context.Context) error {
+ return d.err
+}
+
+func (d *resultDriver) Refresh() {
+}
projects / onehouse / commitdiff