fix(pool): make Destroy honour its own timeout instead of overrunning it

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)
+}

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

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()