From 581f0877f4cb2195e7d1e240f597c05862e1541b Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Sun, 23 Aug 2026 12:14:06 +0400 Subject: [PATCH] fix(pool): make Destroy honour its own timeout instead of overrunning it destroy_timeout was discarded whenever the caller passed a context with a deadline, and on expiry the workers were asked to stop over the relay instead of killed, which adds another 10s. Bound the context by the tighter of the two and kill on expiry. Reset had the same expiry branch. --- pool/static_pool/destroy_test.go | 96 ++++++++++++++++++++++++++++++++ pool/static_pool/pool.go | 8 +-- worker_watcher/worker_watcher.go | 23 +++++--- 3 files changed, 115 insertions(+), 12 deletions(-) create mode 100644 pool/static_pool/destroy_test.go diff --git a/pool/static_pool/destroy_test.go b/pool/static_pool/destroy_test.go new file mode 100644 index 0000000..e2583d5 --- /dev/null +++ b/pool/static_pool/destroy_test.go @@ -0,0 +1,96 @@ +package static_pool + +import ( + "context" + "log/slog" + "os/exec" + "testing" + "time" + + "github.com/roadrunner-server/pool/v2/fsm" + "github.com/roadrunner-server/pool/v2/ipc/pipe" + "github.com/roadrunner-server/pool/v2/payload" + "github.com/roadrunner-server/pool/v2/pool" + "github.com/stretchr/testify/require" +) + +// shutdownSlack covers the one-second ticker in WorkerWatcher plus process teardown. +const shutdownSlack = 3 * time.Second + +// stuckWorkerPool returns a single-worker pool whose worker is parked in a request that never +// finishes, so shutdown can only end by hitting its timeout. +func stuckWorkerPool(t *testing.T, cfg *pool.Config) *Pool { + t.Helper() + + p, err := NewPool( + t.Context(), + func(_ []string) *exec.Cmd { return exec.Command("php", "../../tests/sleep.php") }, + pipe.NewPipeFactory(slog.Default()), + cfg, + slog.Default(), + ) + require.NoError(t, err) + + go func() { + _, _ = p.Exec(context.Background(), &payload.Payload{Body: []byte("hello")}, make(chan struct{})) + }() + + require.Eventually(t, func() bool { + workers := p.Workers() + return len(workers) == 1 && workers[0].State().Compare(fsm.StateWorking) + }, 10*time.Second, 50*time.Millisecond, "worker never started the request") + + return p +} + +func Test_Destroy_KillsStuckWorkerWithinBudget(t *testing.T) { + const budget = time.Second + + for _, tc := range []struct { + name string + destroyTimeout time.Duration + ctxTimeout time.Duration + }{ + {name: "no caller deadline", destroyTimeout: budget}, + {name: "caller deadline is longer", destroyTimeout: budget, ctxTimeout: time.Minute}, + {name: "caller deadline is shorter", destroyTimeout: time.Minute, ctxTimeout: budget}, + } { + t.Run(tc.name, func(t *testing.T) { + p := stuckWorkerPool(t, &pool.Config{ + NumWorkers: 1, + AllocateTimeout: time.Minute, + DestroyTimeout: tc.destroyTimeout, + }) + + ctx := context.Background() + if tc.ctxTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, tc.ctxTimeout) + defer cancel() + } + + start := time.Now() + p.Destroy(ctx) + + require.Less(t, time.Since(start), budget+shutdownSlack) + }) + } +} + +func Test_Reset_KillsStuckWorkerWithinBudget(t *testing.T) { + const budget = time.Second + + p := stuckWorkerPool(t, &pool.Config{ + NumWorkers: 1, + AllocateTimeout: time.Minute, + DestroyTimeout: time.Minute, + ResetTimeout: budget, + }) + t.Cleanup(func() { p.Destroy(context.Background()) }) + + start := time.Now() + require.NoError(t, p.Reset(context.Background())) + + // Reset re-allocates the worker it just killed, so the bound is looser than Destroy's. + require.Less(t, time.Since(start), budget+2*shutdownSlack) +} diff --git a/pool/static_pool/pool.go b/pool/static_pool/pool.go index b8606f8..e275c1f 100644 --- a/pool/static_pool/pool.go +++ b/pool/static_pool/pool.go @@ -162,13 +162,13 @@ func (sp *Pool) RemoveWorker(ctx context.Context) error { return sp.ww.RemoveWorker(ctx) } -// ensureDeadline bounds the context with the fallback timeout when the caller set no deadline. -func ensureDeadline(ctx context.Context, fallback time.Duration) (context.Context, context.CancelFunc) { - if _, ok := ctx.Deadline(); ok { +// ensureDeadline bounds the context by the tighter of its own deadline and the given timeout. +func ensureDeadline(ctx context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { + if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) <= timeout { return ctx, func() {} } - return context.WithTimeout(ctx, fallback) + return context.WithTimeout(ctx, timeout) } // AddWorker adds one worker to the pool. With a dynamic allocator configured, workers above diff --git a/worker_watcher/worker_watcher.go b/worker_watcher/worker_watcher.go index 6ebf591..d4411e4 100644 --- a/worker_watcher/worker_watcher.go +++ b/worker_watcher/worker_watcher.go @@ -245,16 +245,25 @@ func (ww *WorkerWatcher) Release(w *worker.Process) { } } -// stopWatchedWorkers stops every tracked worker concurrently and clears the workers map. -// The caller must hold ww.mu. +// stopWatchedWorkers asks every tracked worker to shut down; Process.Stop waits out its own +// grace period before killing one that does not answer. The caller must hold ww.mu. func (ww *WorkerWatcher) stopWatchedWorkers() { + ww.disposeWatchedWorkers((*worker.Process).Stop) +} + +// killWatchedWorkers terminates every tracked worker at once, for when there is no time left +// to negotiate. The caller must hold ww.mu. +func (ww *WorkerWatcher) killWatchedWorkers() { + ww.disposeWatchedWorkers((*worker.Process).Kill) +} + +func (ww *WorkerWatcher) disposeWatchedWorkers(dispose func(*worker.Process) error) { wg := &sync.WaitGroup{} ww.workers.Range(func(key, value any) bool { w := value.(*worker.Process) wg.Go(func() { w.State().Transition(fsm.StateDestroyed) - // kill the worker - _ = w.Stop() + _ = dispose(w) // remove worker from the channel w.Callback() }) @@ -292,9 +301,8 @@ func (ww *WorkerWatcher) Reset(ctx context.Context) uint64 { return ww.numWorkers.Load() case <-ctx.Done(): - // kill workers ww.mu.Lock() - ww.stopWatchedWorkers() + ww.killWatchedWorkers() ww.container.ResetDone() ww.mu.Unlock() @@ -338,10 +346,9 @@ func (ww *WorkerWatcher) Destroy(ctx context.Context) { ww.mu.Unlock() return case <-ctx.Done(): - // kill workers ww.log.Debug("destroy: context canceled", "error", ctx.Err()) ww.mu.Lock() - ww.stopWatchedWorkers() + ww.killWatchedWorkers() ww.numWorkers.Store(0) ww.destroyed.Store(true) ww.mu.Unlock()