From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 22:14:57 +0000 (-0300) Subject: Run explicit drivers with signal cancellation X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/2015860819c6228bf5ec708a48e5ca9b7c04e27d?ds=sidebyside 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 d38fba4..bffa9bd 100644 --- a/cmd/onehouse/main.go +++ b/cmd/onehouse/main.go @@ -1,13 +1,22 @@ package main import ( - "fmt" + "context" "os" + "os/signal" + "syscall" ) 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 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 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 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 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 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 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 cbf8459..7d23f1d 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -14,7 +14,7 @@ buildGoModule { doInstallCheck = true; installCheckPhase = '' runHook preInstallCheck - [[ "$("$out/bin/onehouse")" == "onehouse" ]] + timeout --preserve-status --signal=TERM 1 "$out/bin/onehouse" runHook postInstallCheck '';