Run explicit drivers with signal cancellation

The runner starts the supplied integrations, forwards reconnect notices,
and waits for each worker to stop. It alone logs failures and preserves
independent errors joined with cancellation. The command waits for a
signal with an empty integration list and completes cleanup before exit.

Issue: 40b314f

diff --git a/cmd/onehouse/main.go b/cmd/onehouse/main.go
index d38fba43fdfd5ac9365b17904dcc1e06c8c27c7c..bffa9bdbd64da723381bedc742fdee496e117835 100644 (file)
--- a/cmd/onehouse/main.go
+++ b/cmd/onehouse/main.go
@@ -1,13 +1,22 @@
 package main
 
 import (
 package main
 
 import (

-       "fmt"
+       "context"

        "os"
        "os"

+       "os/signal"
+       "syscall"

 )
 
 func main() {
 )
 
 func main() {

-       _, err := fmt.Println("onehouse")
-       if err != nil {
-               os.Exit(1)
-       }
+       failed := false
+       defer func() {
+               if failed {
+                       os.Exit(1)
+               }
+       }()
+
+       ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+       defer cancel()
+
+       failed = run(ctx) != nil

 }
 }

diff --git a/cmd/onehouse/run.go b/cmd/onehouse/run.go
new file mode 100644 (file)
index 0000000..0390ac8
--- /dev/null
+++ b/cmd/onehouse/run.go
@@ -0,0 +1,19 @@
+package main
+
+import (
+       "context"
+       "fmt"
+
+       "onehouse/driver"
+)
+
+func run(ctx context.Context) error {
+       integrations := []driver.Driver{}
+
+       err := driver.Run(ctx, integrations, nil)
+       if err != nil {
+               return fmt.Errorf("run integrations: %w", err)
+       }
+
+       return nil
+}

diff --git a/cmd/onehouse/run_test.go b/cmd/onehouse/run_test.go
new file mode 100644 (file)
index 0000000..79cf75c
--- /dev/null
+++ b/cmd/onehouse/run_test.go
@@ -0,0 +1,26 @@
+package main
+
+import (
+       "context"
+       "testing"
+       "testing/synctest"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestRunWaitsForSignalContext(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)
+               }()
+
+               synctest.Wait()
+               require.Empty(t, finished)
+               cancel()
+               require.NoError(t, <-finished)
+       })
+}

diff --git a/doc/driver.rst b/doc/driver.rst
new file mode 100644 (file)
index 0000000..33086f1
--- /dev/null
+++ b/doc/driver.rst
@@ -0,0 +1,16 @@
+##################
+ Driver framework
+##################
+
+``driver.Driver`` supplies ``Run`` and a nonblocking ``Refresh``
+request. The runner starts an explicit list, forwards broker reconnect
+notifications, and waits for cleanup. A driver returns errors without
+logging. The runner logs failures and cancels peers. Cancellation and
+closed transports are expected stops only after the worker context
+stops. A joined independent failure remains a failure.
+
+``cmd/onehouse`` supplies an empty driver list. It waits for SIGINT or
+SIGTERM and finishes cleanup before exit. No integration or broker runs.
+
+Tests use fake drivers. They do not verify a broker or a physical
+device.

diff --git a/driver/driver.go b/driver/driver.go
new file mode 100644 (file)
index 0000000..d6a1901
--- /dev/null
+++ b/driver/driver.go
@@ -0,0 +1,11 @@
+package driver
+
+import "context"
+
+// Driver owns one integration. Run returns its failures without logging and
+// stops on cancellation. Refresh requests a snapshot after broker reconnect.
+// Refresh must be safe during Run and must not block the caller.
+type Driver interface {
+       Run(ctx context.Context) error
+       Refresh()
+}

diff --git a/driver/runner.go b/driver/runner.go
new file mode 100644 (file)
index 0000000..e918416
--- /dev/null
+++ b/driver/runner.go
@@ -0,0 +1,99 @@
+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)
+}

diff --git a/driver/runner_test.go b/driver/runner_test.go
new file mode 100644 (file)
index 0000000..2ece6cd
--- /dev/null
+++ b/driver/runner_test.go
@@ -0,0 +1,160 @@
+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() {
+}

diff --git a/nix/package.nix b/nix/package.nix
index cbf84591d36df6532ba4b1a1eedf20650aee2b69..7d23f1dfd7f7a7e6463854869a7e20eb7d6bda9c 100644 (file)
--- a/nix/package.nix
+++ b/nix/package.nix
@@ -14,7 +14,7 @@ buildGoModule {
   doInstallCheck = true;
   installCheckPhase = ''
     runHook preInstallCheck
   doInstallCheck = true;
   installCheckPhase = ''
     runHook preInstallCheck

-    [[ "$("$out/bin/onehouse")" == "onehouse" ]]
+    timeout --preserve-status --signal=TERM 1 "$out/bin/onehouse"

     runHook postInstallCheck
   '';
 
     runHook postInstallCheck
   '';