fix(backend): bound the launcher health probe with a timeout (#2017)
http.Get uses http.DefaultClient, which has no timeout and carries no context. A backend that accepted the TCP connection but never responded blocked waitForHealth forever, hanging the launcher with no way to cancel it -- exactly what the suppressed noctx lint was warning about. Probe through a package-level client with an explicit 2s timeout using http.NewRequestWithContext, thread a context down from launcher.Run, and drain each response body before closing so the connection is reused across poll iterations. The overall wait now derives a deadline from the caller's context, so cancellation ends it too. Co-authored-by: Kandev Agent <agent@kandev.dev> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,13 +1,26 @@
|
||||
package launcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
healthPollInterval = 300 * time.Millisecond
|
||||
healthProbeTimeout = 2 * time.Second
|
||||
)
|
||||
|
||||
// healthProbeClient bounds every individual health request. http.DefaultClient
|
||||
// has no timeout, so a backend that accepts the connection but never answers
|
||||
// would otherwise block the launcher forever.
|
||||
var healthProbeClient = &http.Client{Timeout: healthProbeTimeout}
|
||||
|
||||
type childState interface {
|
||||
Exited() (bool, int)
|
||||
}
|
||||
@@ -24,27 +37,47 @@ func healthTimeout(defaultMS int) time.Duration {
|
||||
return time.Duration(n) * time.Millisecond
|
||||
}
|
||||
|
||||
func waitForHealth(baseURL string, proc childState, timeout time.Duration, onFailure func()) error {
|
||||
deadline := time.Now().Add(timeout)
|
||||
func waitForHealth(ctx context.Context, baseURL string, proc childState, timeout time.Duration, onFailure func()) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
healthURL := baseURL + "/health"
|
||||
for time.Now().Before(deadline) {
|
||||
for ctx.Err() == nil {
|
||||
if exited, code := proc.Exited(); exited {
|
||||
if onFailure != nil {
|
||||
onFailure()
|
||||
}
|
||||
return fmt.Errorf("backend exited (code %d) before healthcheck passed", code)
|
||||
}
|
||||
resp, err := http.Get(healthURL) //nolint:gosec,noctx
|
||||
if err == nil {
|
||||
_ = resp.Body.Close()
|
||||
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||||
return nil
|
||||
}
|
||||
if probeHealth(ctx, healthURL) {
|
||||
return nil
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(healthPollInterval):
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
if onFailure != nil {
|
||||
onFailure()
|
||||
}
|
||||
if errors.Is(ctx.Err(), context.Canceled) {
|
||||
return fmt.Errorf("backend healthcheck canceled at %s: %w", healthURL, ctx.Err())
|
||||
}
|
||||
return fmt.Errorf("backend healthcheck timed out after %s at %s", timeout, healthURL)
|
||||
}
|
||||
|
||||
// probeHealth reports whether a single health request succeeded. The body is
|
||||
// drained and closed so the connection can be reused by the next poll.
|
||||
func probeHealth(ctx context.Context, healthURL string) bool {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthURL, nil)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
resp, err := healthProbeClient.Do(req)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
return resp.StatusCode >= 200 && resp.StatusCode < 300
|
||||
}
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
package launcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type fakeChild struct {
|
||||
exited bool
|
||||
code int
|
||||
}
|
||||
|
||||
func (f fakeChild) Exited() (bool, int) { return f.exited, f.code }
|
||||
|
||||
func TestHealthTimeoutUsesDefaultWhenEnvUnusable(t *testing.T) {
|
||||
for name, raw := range map[string]string{
|
||||
"unset": "",
|
||||
"not-numeric": "soon",
|
||||
"zero": "0",
|
||||
"negative": "-5",
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
t.Setenv("KANDEV_HEALTH_TIMEOUT_MS", raw)
|
||||
if got := healthTimeout(1500); got != 1500*time.Millisecond {
|
||||
t.Fatalf("healthTimeout() = %s, want 1.5s", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHealthTimeoutHonorsEnvOverride(t *testing.T) {
|
||||
t.Setenv("KANDEV_HEALTH_TIMEOUT_MS", "250")
|
||||
if got := healthTimeout(1500); got != 250*time.Millisecond {
|
||||
t.Fatalf("healthTimeout() = %s, want 250ms", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitForHealthReturnsNilOnHealthyResponse(t *testing.T) {
|
||||
var calls atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/health" {
|
||||
t.Errorf("path = %q, want /health", r.URL.Path)
|
||||
}
|
||||
// The first probe is unhealthy so the poll loop iterates at least once,
|
||||
// exercising body drain + connection reuse across iterations.
|
||||
if calls.Add(1) == 1 {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
_, _ = w.Write([]byte("starting"))
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte("ok"))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
var failures int
|
||||
err := waitForHealth(context.Background(), srv.URL, fakeChild{}, 5*time.Second, func() { failures++ })
|
||||
if err != nil {
|
||||
t.Fatalf("waitForHealth() = %v, want nil", err)
|
||||
}
|
||||
if failures != 0 {
|
||||
t.Fatalf("onFailure called %d times, want 0", failures)
|
||||
}
|
||||
if got := calls.Load(); got < 2 {
|
||||
t.Fatalf("probe calls = %d, want at least 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitForHealthTimesOutOnNon2xxResponse(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
var failures int
|
||||
err := waitForHealth(context.Background(), srv.URL, fakeChild{}, 400*time.Millisecond, func() { failures++ })
|
||||
if err == nil {
|
||||
t.Fatal("waitForHealth() = nil, want timeout error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "timed out") {
|
||||
t.Fatalf("error = %v, want a timeout error", err)
|
||||
}
|
||||
if failures != 1 {
|
||||
t.Fatalf("onFailure called %d times, want 1", failures)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitForHealthReturnsWhenServerHangs(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
|
||||
// Accept the connection but never respond, the failure mode that made
|
||||
// the unbounded http.DefaultClient block the launcher forever.
|
||||
select {
|
||||
case <-release:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
}))
|
||||
t.Cleanup(func() {
|
||||
close(release)
|
||||
srv.Close()
|
||||
})
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- waitForHealth(context.Background(), srv.URL, fakeChild{}, 300*time.Millisecond, nil)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if err == nil {
|
||||
t.Fatal("waitForHealth() = nil, want timeout error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "timed out") {
|
||||
t.Fatalf("error = %v, want a timeout error", err)
|
||||
}
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("waitForHealth blocked on a hanging server instead of timing out")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitForHealthReturnsWhenContextCanceled(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
|
||||
select {
|
||||
case <-release:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
}))
|
||||
t.Cleanup(func() {
|
||||
close(release)
|
||||
srv.Close()
|
||||
})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
// A generous timeout: only cancellation can end this wait in time.
|
||||
done <- waitForHealth(ctx, srv.URL, fakeChild{}, time.Minute, nil)
|
||||
}()
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if err == nil {
|
||||
t.Fatal("waitForHealth() = nil, want cancellation error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "canceled") {
|
||||
t.Fatalf("error = %v, want a cancellation error", err)
|
||||
}
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("waitForHealth ignored context cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitForHealthFailsFastWhenBackendExited(t *testing.T) {
|
||||
var failures int
|
||||
err := waitForHealth(
|
||||
context.Background(),
|
||||
"http://127.0.0.1:1",
|
||||
fakeChild{exited: true, code: 3},
|
||||
time.Minute,
|
||||
func() { failures++ },
|
||||
)
|
||||
if err == nil {
|
||||
t.Fatal("waitForHealth() = nil, want exit error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "code 3") {
|
||||
t.Fatalf("error = %v, want the backend exit code", err)
|
||||
}
|
||||
if failures != 1 {
|
||||
t.Fatalf("onFailure called %d times, want 1", failures)
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package launcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
@@ -30,11 +31,14 @@ func Run(args []string, build BuildInfo) int {
|
||||
fmt.Print(cli.Help())
|
||||
return 0
|
||||
}
|
||||
// Root context for the launcher process; bounds the backend health probe so
|
||||
// an unresponsive backend cannot hang the launcher indefinitely.
|
||||
ctx := context.Background()
|
||||
switch opts.Command {
|
||||
case CommandStart:
|
||||
return runStart(opts)
|
||||
return runStart(ctx, opts)
|
||||
case CommandRun:
|
||||
return runInstalled(opts)
|
||||
return runInstalled(ctx, opts)
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "[kandev] native launcher command %q is not implemented yet\n", opts.Command)
|
||||
return 1
|
||||
|
||||
@@ -1,12 +1,13 @@
|
||||
package launcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
func runInstalled(opts Options) int {
|
||||
func runInstalled(ctx context.Context, opts Options) int {
|
||||
backendPort, err := resolvePorts(opts)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "[kandev] "+err.Error())
|
||||
@@ -32,7 +33,7 @@ func runInstalled(opts Options) int {
|
||||
if releaseTag == "" {
|
||||
releaseTag = "(" + bundle.Source + ")"
|
||||
}
|
||||
return launchManaged(managedAppConfig{
|
||||
return launchManaged(ctx, managedAppConfig{
|
||||
Header: "release: " + releaseTag,
|
||||
Mode: "run",
|
||||
Backend: bundle.Launcher,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package launcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -19,7 +20,7 @@ var (
|
||||
}
|
||||
)
|
||||
|
||||
func runStart(opts Options) int {
|
||||
func runStart(ctx context.Context, opts Options) int {
|
||||
backendPort, err := resolvePorts(opts)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "[kandev] "+err.Error())
|
||||
@@ -42,7 +43,7 @@ func runStart(opts Options) int {
|
||||
fmt.Fprintln(os.Stderr, "[kandev] "+err.Error())
|
||||
return 1
|
||||
}
|
||||
return launchManaged(managedAppConfig{
|
||||
return launchManaged(ctx, managedAppConfig{
|
||||
Header: "start mode: using local build",
|
||||
Mode: "start",
|
||||
Backend: self,
|
||||
@@ -77,7 +78,7 @@ func resolveLogLevel(opts Options) string {
|
||||
}
|
||||
}
|
||||
|
||||
func runManagedApp(cfg managedAppConfig) int {
|
||||
func runManagedApp(ctx context.Context, cfg managedAppConfig) int {
|
||||
ignoreBrokenPipeSignal()
|
||||
logStartup(cfg.Header, cfg.Ports, resolveDatabasePath(), cfg.LogLevel)
|
||||
setLauncherShutdownDebug(cfg.Opts.Debug || os.Getenv("KANDEV_SHUTDOWN_DEBUG") == "1")
|
||||
@@ -94,7 +95,7 @@ func runManagedApp(cfg managedAppConfig) int {
|
||||
}
|
||||
shutdownDebugf("runManagedApp backend launched")
|
||||
fmt.Println("[kandev] starting backend...")
|
||||
if err := waitForHealthFn(cfg.Ports.BackendURL, backend, healthTimeout(healthTimeoutReleaseMS), dumpLogs); err != nil {
|
||||
if err := waitForHealthFn(ctx, cfg.Ports.BackendURL, backend, healthTimeout(healthTimeoutReleaseMS), dumpLogs); err != nil {
|
||||
supervisor.shutdown("backend health failure")
|
||||
fmt.Fprintln(os.Stderr, "[kandev] "+err.Error())
|
||||
return 1
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package launcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"testing"
|
||||
@@ -21,13 +22,13 @@ func TestRunStartUsesSelfExecutableAndBackendCWD(t *testing.T) {
|
||||
}
|
||||
|
||||
var got managedAppConfig
|
||||
launchManaged = func(cfg managedAppConfig) int {
|
||||
launchManaged = func(_ context.Context, cfg managedAppConfig) int {
|
||||
got = cfg
|
||||
return 42
|
||||
}
|
||||
t.Setenv("KANDEV_HOME_DIR", t.TempDir())
|
||||
|
||||
code := runStart(Options{Command: CommandStart, BackendPort: 48123, Headless: true})
|
||||
code := runStart(context.Background(), Options{Command: CommandStart, BackendPort: 48123, Headless: true})
|
||||
if code != 42 {
|
||||
t.Fatalf("runStart() = %d, want 42", code)
|
||||
}
|
||||
@@ -83,13 +84,13 @@ func TestRunManagedAppAttachesSignalsBeforeBackendLaunch(t *testing.T) {
|
||||
attachSignalsFn = func(_ *processSupervisor) {
|
||||
events = append(events, "attach-signals")
|
||||
}
|
||||
waitForHealthFn = func(_ string, _ childState, _ time.Duration, _ func()) error {
|
||||
waitForHealthFn = func(_ context.Context, _ string, _ childState, _ time.Duration, _ func()) error {
|
||||
events = append(events, "wait-health")
|
||||
return nil
|
||||
}
|
||||
t.Setenv("KANDEV_HOME_DIR", t.TempDir())
|
||||
|
||||
code := runManagedApp(managedAppConfig{
|
||||
code := runManagedApp(context.Background(), managedAppConfig{
|
||||
Header: "test",
|
||||
Mode: "start",
|
||||
Backend: "kandev",
|
||||
|
||||
Reference in New Issue
Block a user