feat: supervise module lifecycle and managed workers with bounded shutdown #13

Merged
end merged 6 commits from feat/lifecycle into main 2026-10-27 15:01:51 -07:00
2 changed files with 90 additions and 30 deletions
Showing only changes of commit f4abb626c6 - Show all commits

No files matched your search

+8 -9
View File
@@ -238,7 +238,7 @@ func (a *App) Run(parent context.Context) error {
if !ok {
signals = nil
} else if err := receiveSignal(sig); err != nil {
return a.result(err)
return a.result(parent, err)
}
continue
default:
@@ -246,7 +246,6 @@ func (a *App) Run(parent context.Context) error {
a.mu.Lock()
if !a.stopping && parent.Err() != nil {
a.errs = append(a.errs, parent.Err())
a.stopLocked()
parentDone = nil
}
@@ -288,7 +287,7 @@ func (a *App) Run(parent context.Context) error {
if stopping {
if err := shutdown.Err(); err != nil {
return a.result(err)
return a.result(parent, err)
}
if setupDone == nil && teardownDone == nil {
if teardownIndex >= 0 {
@@ -297,7 +296,7 @@ func (a *App) Run(parent context.Context) error {
teardownDone = result
go func() { result <- m.Teardown(shutdown) }()
} else if active == 0 {
return a.result(nil)
return a.result(parent, nil)
}
}
}
@@ -325,7 +324,6 @@ func (a *App) Run(parent context.Context) error {
case <-a.wake:
case <-parentDone:
a.mu.Lock()
a.errs = append(a.errs, parent.Err())
a.stopLocked()
a.mu.Unlock()
parentDone = nil
@@ -335,16 +333,17 @@ func (a *App) Run(parent context.Context) error {
continue
}
if err := receiveSignal(sig); err != nil {
return a.result(err)
return a.result(parent, err)
}
case <-deadline:
return a.result(context.DeadlineExceeded)
return a.result(parent, context.DeadlineExceeded)
}
}
}
func (a *App) result(extra error) error {
func (a *App) result(parent context.Context, extra error) error {
a.mu.Lock()
defer a.mu.Unlock()
return errors.Join(append(a.errs, extra)...)
// invariant: every completion and abandonment retains parent cancellation exactly once.
return errors.Join(append(a.errs, parent.Err(), extra)...)
}
+82 -21
View File
@@ -380,28 +380,89 @@ func TestAdmissionRace(t *testing.T) {
}
func TestIndependentDescendantCancellation(t *testing.T) {
signals := make(chan os.Signal, 2)
a := newApp(t, signals, time.Second)
canceled := make(chan struct{})
allowReturn := make(chan struct{})
add(t, a, Module{Name: "owner", Setup: func(ctx context.Context) error {
child, cancel := context.WithCancel(ctx)
err := a.Go(child, "independent", func(ctx context.Context) error {
cancel()
close(canceled)
<-allowReturn
return fmt.Errorf("worker result: %w", errors.Join(ctx.Err(), errors.New("other")))
for _, mixed := range []bool{false, true} {
t.Run(map[bool]string{false: "cancellation-only", true: "mixed"}[mixed], func(t *testing.T) {
signals := make(chan os.Signal, 2)
a := newApp(t, signals, time.Second)
canceled := make(chan struct{})
allowReturn := make(chan struct{})
other := errors.New("other")
add(t, a, Module{Name: "owner", Setup: func(ctx context.Context) error {
child, cancel := context.WithCancel(ctx)
err := a.Go(child, "independent", func(ctx context.Context) error {
cancel()
close(canceled)
<-allowReturn
err := ctx.Err()
if mixed {
err = errors.Join(err, other)
}
return fmt.Errorf("worker result: %w", err)
})
if err != nil {
cancel()
}
return err
}, Teardown: func(context.Context) error { close(allowReturn); return nil }})
done := run(a, context.Background())
wait(t, canceled)
signals <- syscall.SIGTERM
if err := receive(t, done); !errors.Is(err, context.Canceled) || errors.Is(err, other) != mixed {
t.Fatalf("independent cancellation lost: %v", err)
}
})
}
}
func TestParentCancellationDuringShutdown(t *testing.T) {
for _, boundary := range []string{"completion", "deadline", "second-signal"} {
t.Run(boundary, func(t *testing.T) {
iterations := 1
if boundary == "completion" {
iterations = 1000
}
for range iterations {
parent, cancel := context.WithCancel(context.Background())
signals := make(chan os.Signal, 2)
timeout := time.Second
if boundary == "deadline" {
timeout = 20 * time.Millisecond
}
a := newApp(t, signals, timeout)
exited := make(chan struct{})
add(t, a, Module{Name: "last", Teardown: func(ctx context.Context) error {
defer close(exited)
cancel()
switch boundary {
case "deadline":
<-ctx.Done()
case "second-signal":
signals <- syscall.SIGINT
signals <- syscall.SIGTERM
<-ctx.Done()
}
return nil
}})
err := a.Run(parent)
cancel()
wait(t, exited)
if !errors.Is(err, context.Canceled) {
t.Fatalf("parent cancellation lost: %v", err)
}
if errors.Is(err, context.DeadlineExceeded) != (boundary == "deadline") {
t.Fatalf("unexpected shutdown result: %v", err)
}
parentErrors := 0
for _, child := range err.(interface{ Unwrap() []error }).Unwrap() {
if child == context.Canceled {
parentErrors++
}
}
if parentErrors != 1 {
t.Fatalf("parent error recorded %d times: %v", parentErrors, err)
}
}
})
if err != nil {
cancel()
}
return err
}, Teardown: func(context.Context) error { close(allowReturn); return nil }})
done := run(a, context.Background())
wait(t, canceled)
signals <- syscall.SIGTERM
if err := receive(t, done); !errors.Is(err, context.Canceled) || !strings.Contains(err.Error(), "other") {
t.Fatalf("independent cancellation lost: %v", err)
}
}