From 0b5c210396568ff484081639d526abd3189680c0 Mon Sep 17 00:00:00 2001 From: Martijn Date: Thu, 12 Mar 2026 10:17:32 +0100 Subject: [PATCH 01/17] Add support for review apps or scale-to-zero configurations --- internal/cmd/deploy.go | 3 + internal/cmd/run.go | 3 +- internal/server/config.go | 2 + internal/server/docker_client.go | 64 ++++++ internal/server/docker_client_test.go | 44 +++++ internal/server/idle_controller.go | 252 ++++++++++++++++++++++++ internal/server/idle_controller_test.go | 75 +++++++ internal/server/router.go | 29 ++- internal/server/router_test.go | 6 +- internal/server/service.go | 93 ++++++++- internal/server/service_test.go | 2 +- internal/server/testing.go | 2 +- 12 files changed, 559 insertions(+), 16 deletions(-) create mode 100644 internal/server/docker_client.go create mode 100644 internal/server/docker_client_test.go create mode 100644 internal/server/idle_controller.go create mode 100644 internal/server/idle_controller_test.go diff --git a/internal/cmd/deploy.go b/internal/cmd/deploy.go index 82af0e18..cfc5e2da 100644 --- a/internal/cmd/deploy.go +++ b/internal/cmd/deploy.go @@ -52,6 +52,9 @@ func newDeployCommand() *deployCommand { deployCommand.cmd.Flags().DurationVar(&deployCommand.args.ServiceOptions.WriterAffinityTimeout, "writer-affinity-timeout", server.DefaultWriterAffinityTimeout, "Time after a write before read requests will be routed to readers") deployCommand.cmd.Flags().BoolVar(&deployCommand.args.ServiceOptions.ReadTargetsAcceptWebsockets, "read-target-websockets", false, "Route WebSocket traffic to read targets, when available") + deployCommand.cmd.Flags().DurationVar(&deployCommand.args.ServiceOptions.IdleTimeout, "idle-timeout", 0, "Stop container after this duration of inactivity (0 to disable)") + deployCommand.cmd.Flags().DurationVar(&deployCommand.args.ServiceOptions.IdleWakeTimeout, "idle-wake-timeout", server.DefaultIdleWakeTimeout, "Max time to hold request while waking container") + deployCommand.cmd.Flags().DurationVar(&deployCommand.args.TargetOptions.ResponseTimeout, "target-timeout", server.DefaultTargetTimeout, "Maximum time to wait for the target server to respond when serving requests") deployCommand.cmd.Flags().BoolVar(&deployCommand.args.TargetOptions.BufferRequests, "buffer-requests", false, "Buffer requests before forwarding to target") diff --git a/internal/cmd/run.go b/internal/cmd/run.go index 0a70655e..1c2689ea 100644 --- a/internal/cmd/run.go +++ b/internal/cmd/run.go @@ -29,6 +29,7 @@ func newRunCommand() *runCommand { runCommand.cmd.Flags().IntVar(&globalConfig.HttpsPort, "https-port", getEnvInt("HTTPS_PORT", server.DefaultHttpsPort), "Port to serve HTTPS traffic on") runCommand.cmd.Flags().IntVar(&globalConfig.MetricsPort, "metrics-port", getEnvInt("METRICS_PORT", 0), "Publish metrics on the specified port (default zero to disable)") runCommand.cmd.Flags().BoolVar(&globalConfig.HTTP3Enabled, "http3", false, "Enable HTTP/3") + runCommand.cmd.Flags().StringVar(&globalConfig.DockerSocketPath, "docker-socket", getEnvString("DOCKER_SOCKET", server.DefaultDockerSocketPath), "Path to Docker socket") return runCommand } @@ -36,7 +37,7 @@ func newRunCommand() *runCommand { func (c *runCommand) run(cmd *cobra.Command, args []string) error { c.setLogger() - router := server.NewRouter(globalConfig.StatePath()) + router := server.NewRouter(globalConfig.StatePath(), globalConfig.DockerSocketPath) router.RestoreLastSavedState() s := server.NewServer(&globalConfig, router) diff --git a/internal/server/config.go b/internal/server/config.go index 66f632a2..052b069a 100644 --- a/internal/server/config.go +++ b/internal/server/config.go @@ -10,6 +10,7 @@ import ( const ( DefaultHttpPort = 80 DefaultHttpsPort = 443 + DefaultDockerSocketPath = "/var/run/docker.sock" ) type Config struct { @@ -20,6 +21,7 @@ type Config struct { HTTP3Enabled bool AlternateConfigDir string + DockerSocketPath string } func (c Config) SocketPath() string { diff --git a/internal/server/docker_client.go b/internal/server/docker_client.go new file mode 100644 index 00000000..fc087faf --- /dev/null +++ b/internal/server/docker_client.go @@ -0,0 +1,64 @@ +package server + +import ( + "context" + "fmt" + "net" + "net/http" +) + +type DockerClient struct { + httpClient *http.Client +} + +func NewDockerClient(socketPath string) *DockerClient { + return &DockerClient{ + httpClient: &http.Client{ + Transport: &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, "unix", socketPath) + }, + }, + }, + } +} + +func (c *DockerClient) StopContainer(ctx context.Context, name string) error { + url := fmt.Sprintf("http://localhost/v1.41/containers/%s/stop", name) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + return err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotModified { + return fmt.Errorf("unexpected status code from docker stop: %d", resp.StatusCode) + } + + return nil +} + +func (c *DockerClient) StartContainer(ctx context.Context, name string) error { + url := fmt.Sprintf("http://localhost/v1.41/containers/%s/start", name) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + return err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotModified { + return fmt.Errorf("unexpected status code from docker start: %d", resp.StatusCode) + } + + return nil +} diff --git a/internal/server/docker_client_test.go b/internal/server/docker_client_test.go new file mode 100644 index 00000000..e3c5b4c6 --- /dev/null +++ b/internal/server/docker_client_test.go @@ -0,0 +1,44 @@ +package server + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestDockerClient_StopStart(t *testing.T) { + stopCalled := false + startCalled := false + + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1.41/containers/test-container/stop" { + stopCalled = true + w.WriteHeader(http.StatusNoContent) + } else if r.URL.Path == "/v1.41/containers/test-container/start" { + startCalled = true + w.WriteHeader(http.StatusNoContent) + } + })) + + socketPath := t.TempDir() + "/docker.sock" + l, err := net.Listen("unix", socketPath) + require.NoError(t, err) + server.Listener = l + server.Start() + defer server.Close() + + client := NewDockerClient(socketPath) + + err = client.StopContainer(context.Background(), "test-container") + assert.NoError(t, err) + assert.True(t, stopCalled) + + err = client.StartContainer(context.Background(), "test-container") + assert.NoError(t, err) + assert.True(t, startCalled) +} diff --git a/internal/server/idle_controller.go b/internal/server/idle_controller.go new file mode 100644 index 00000000..7486aa6a --- /dev/null +++ b/internal/server/idle_controller.go @@ -0,0 +1,252 @@ +package server + +import ( + "context" + "encoding/json" + "log/slog" + "sync" + "time" +) + +type IdleState int + +const ( + IdleStateActive IdleState = iota + IdleStateSleeping + IdleStateWaking +) + +func (is IdleState) String() string { + switch is { + case IdleStateActive: + return "active" + case IdleStateSleeping: + return "sleeping" + case IdleStateWaking: + return "waking" + default: + return "" + } +} + +type IdleWaitAction int + +const ( + IdleWaitActionProceed IdleWaitAction = iota + IdleWaitActionTimedOut +) + +type IdleController struct { + State IdleState `json:"state"` + IdleTimeout time.Duration `json:"idle_timeout"` + WakeTimeout time.Duration `json:"wake_timeout"` + ContainerNames []string `json:"container_names"` + + serviceName string + docker *DockerClient + lb *LoadBalancer + + lock sync.RWMutex + lastRequestAt time.Time + wakeChan chan bool + closeChan chan bool + disabled bool +} + +func NewIdleController(serviceName string, idleTimeout, wakeTimeout time.Duration, containerNames []string, docker *DockerClient, lb *LoadBalancer) *IdleController { + ic := &IdleController{ + State: IdleStateActive, + IdleTimeout: idleTimeout, + WakeTimeout: wakeTimeout, + ContainerNames: containerNames, + serviceName: serviceName, + docker: docker, + lb: lb, + lastRequestAt: time.Now(), + closeChan: make(chan bool), + } + + go ic.run() + return ic +} + +func (ic *IdleController) UnmarshalJSON(data []byte) error { + type alias IdleController + aux := &struct { + *alias + }{ + alias: (*alias)(ic), + } + if err := json.Unmarshal(data, &aux); err != nil { + return err + } + + ic.lastRequestAt = time.Now() + ic.closeChan = make(chan bool) + + go ic.run() + return nil +} + +func (ic *IdleController) TrackActivity() { + ic.lock.Lock() + defer ic.lock.Unlock() + + ic.lastRequestAt = time.Now() +} + +func (ic *IdleController) GetState() IdleState { + ic.lock.RLock() + defer ic.lock.RUnlock() + return ic.State +} + +func (ic *IdleController) WaitIfSleeping() IdleWaitAction { + ic.lock.RLock() + state := ic.State + wakeChan := ic.wakeChan + wakeTimeout := ic.WakeTimeout + ic.lock.RUnlock() + + if state == IdleStateActive { + return IdleWaitActionProceed + } + + if state == IdleStateSleeping { + ic.wake() + // Re-read wakeChan + ic.lock.RLock() + wakeChan = ic.wakeChan + ic.lock.RUnlock() + } + + if wakeChan == nil { + return IdleWaitActionProceed + } + + select { + case <-wakeChan: + return IdleWaitActionProceed + case <-time.After(wakeTimeout): + return IdleWaitActionTimedOut + } +} + +func (ic *IdleController) UpdateContainers(names []string) { + ic.lock.Lock() + defer ic.lock.Unlock() + + ic.ContainerNames = names + ic.lastRequestAt = time.Now() + + if ic.State != IdleStateActive { + ic.State = IdleStateActive + if ic.wakeChan != nil { + close(ic.wakeChan) + ic.wakeChan = nil + } + } +} + +func (ic *IdleController) Disable() { + ic.lock.Lock() + defer ic.lock.Unlock() + ic.disabled = true +} + +func (ic *IdleController) Enable() { + ic.lock.Lock() + defer ic.lock.Unlock() + ic.disabled = false + ic.lastRequestAt = time.Now() +} + +func (ic *IdleController) Close() { + close(ic.closeChan) +} + +func (ic *IdleController) run() { + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ic.closeChan: + return + case <-ticker.C: + ic.checkIdle() + } + } +} + +func (ic *IdleController) checkIdle() { + ic.lock.Lock() + if ic.disabled || ic.State != IdleStateActive || ic.IdleTimeout <= 0 { + ic.lock.Unlock() + return + } + + if time.Since(ic.lastRequestAt) > ic.IdleTimeout { + ic.State = IdleStateSleeping + ic.wakeChan = make(chan bool) + containerNames := ic.ContainerNames + ic.lock.Unlock() + + slog.Info("Service is idle, stopping containers", "service", ic.serviceName, "containers", containerNames) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + for _, name := range containerNames { + if err := ic.docker.StopContainer(ctx, name); err != nil { + slog.Error("Failed to stop idle container", "service", ic.serviceName, "container", name, "error", err) + } + } + } else { + ic.lock.Unlock() + } +} + +func (ic *IdleController) wake() { + ic.lock.Lock() + if ic.State != IdleStateSleeping { + ic.lock.Unlock() + return + } + + ic.State = IdleStateWaking + containerNames := ic.ContainerNames + ic.lock.Unlock() + + slog.Info("Service waking up, starting containers", "service", ic.serviceName, "containers", containerNames) + + go func() { + ctx, cancel := context.WithTimeout(context.Background(), ic.WakeTimeout) + defer cancel() + + var wg sync.WaitGroup + wg.Add(len(containerNames)) + for _, name := range containerNames { + go func(n string) { + defer wg.Done() + if err := ic.docker.StartContainer(ctx, n); err != nil { + slog.Error("Failed to start container during wake", "service", ic.serviceName, "container", n, "error", err) + } + }(name) + } + wg.Wait() + + // Wait until healthy + err := ic.lb.WaitUntilHealthy(ic.WakeTimeout) + if err != nil { + slog.Error("Service failed to become healthy after wake", "service", ic.serviceName, "error", err) + } + + ic.lock.Lock() + ic.State = IdleStateActive + ic.lastRequestAt = time.Now() + close(ic.wakeChan) + ic.wakeChan = nil + ic.lock.Unlock() + }() +} diff --git a/internal/server/idle_controller_test.go b/internal/server/idle_controller_test.go new file mode 100644 index 00000000..11dc5215 --- /dev/null +++ b/internal/server/idle_controller_test.go @@ -0,0 +1,75 @@ +package server + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestIdleController_IdleAndWake(t *testing.T) { + stopCalled := make(chan bool, 1) + startCalled := make(chan bool, 1) + + dockerServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1.41/containers/test-container/stop" { + stopCalled <- true + w.WriteHeader(http.StatusNoContent) + } else if r.URL.Path == "/v1.41/containers/test-container/start" { + startCalled <- true + w.WriteHeader(http.StatusNoContent) + } + })) + + socketPath := t.TempDir() + "/docker.sock" + l, err := net.Listen("unix", socketPath) + require.NoError(t, err) + dockerServer.Listener = l + dockerServer.Start() + defer dockerServer.Close() + + dockerClient := NewDockerClient(socketPath) + + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) + })) + defer backend.Close() + + target, _ := NewTarget(backend.URL[7:], defaultTargetOptions) + tl := TargetList{target} + lb := NewLoadBalancer(tl, 0, false) + lb.MarkAllHealthy() + + ic := NewIdleController("test", 100*time.Millisecond, time.Second, []string{"test-container"}, dockerClient, lb) + defer ic.Close() + + // Initial state: active + assert.Equal(t, IdleStateActive, ic.GetState()) + + // Wait for idle to trigger + select { + case <-stopCalled: + case <-time.After(time.Second): + t.Fatal("Timeout waiting for container to be stopped") + } + + assert.Equal(t, IdleStateSleeping, ic.GetState()) + + // Wake up on request + action := ic.WaitIfSleeping() + assert.Equal(t, IdleWaitActionProceed, action) + + select { + case <-startCalled: + case <-time.After(time.Second): + t.Fatal("Timeout waiting for container to be started") + } + + assert.Equal(t, IdleStateActive, ic.GetState()) +} diff --git a/internal/server/router.go b/internal/server/router.go index 3b891b6a..792e8df8 100644 --- a/internal/server/router.go +++ b/internal/server/router.go @@ -47,9 +47,10 @@ func RoutedTargetPath(r *http.Request) string { } type Router struct { - statePath string - services *ServiceMap - serviceLock sync.RWMutex + statePath string + dockerClient *DockerClient + services *ServiceMap + serviceLock sync.RWMutex } type ServiceDescription struct { @@ -62,10 +63,11 @@ type ServiceDescription struct { type ServiceDescriptionMap map[string]ServiceDescription -func NewRouter(statePath string) *Router { +func NewRouter(statePath string, dockerSocketPath string) *Router { return &Router{ - statePath: statePath, - services: NewServiceMap(), + statePath: statePath, + dockerClient: NewDockerClient(dockerSocketPath), + services: NewServiceMap(), } } @@ -91,6 +93,8 @@ func (r *Router) RestoreLastSavedState() error { r.withWriteLock(func() error { r.services = NewServiceMap() for _, service := range services { + service.docker = r.dockerClient + service.initialize(service.options, service.targetOptions) r.services.Set(service) } @@ -259,12 +263,20 @@ func (r *Router) ListActiveServices() ServiceDescriptionMap { path := strings.Join(service.options.PathPrefixes, ",") target := strings.Join(service.active.Targets().Names(), ",") + state := service.pauseController.GetState().String() + if service.idleController != nil && service.pauseController.GetState() == PauseStateRunning { + icState := service.idleController.GetState() + if icState != IdleStateActive { + state = icState.String() + } + } + result[name] = ServiceDescription{ Host: host, Path: path, Target: target, TLS: service.options.TLSEnabled, - State: service.pauseController.GetState().String(), + State: state, } } } @@ -305,9 +317,10 @@ func (r *Router) GetCertificate(hello *tls.ClientHelloInfo) (*tls.Certificate, e func (r *Router) createOrUpdateService(name string, options ServiceOptions, targetOptions TargetOptions) (*Service, error) { service := r.services.Get(name) if service == nil { - return NewService(name, options, targetOptions) + return NewService(name, options, targetOptions, r.dockerClient) } + service.docker = r.dockerClient err := service.UpdateOptions(options, targetOptions) return service, err } diff --git a/internal/server/router_test.go b/internal/server/router_test.go index 28184868..650e91ae 100644 --- a/internal/server/router_test.go +++ b/internal/server/router_test.go @@ -767,7 +767,7 @@ func TestRouter_RestoreLastSavedState(t *testing.T) { _, second := testBackend(t, "second", http.StatusOK) _, third := testBackend(t, "third", http.StatusOK) - router := NewRouter(statePath) + router := NewRouter(statePath, DefaultDockerSocketPath) require.NoError(t, router.DeployService("default", []string{first}, defaultEmptyReaders, defaultServiceOptions, defaultTargetOptions, defaultDeploymentOptions)) serviceOptions := defaultServiceOptions @@ -791,7 +791,7 @@ func TestRouter_RestoreLastSavedState(t *testing.T) { assert.Equal(t, http.StatusOK, statusCode) assert.Equal(t, "third", body) - router = NewRouter(statePath) + router = NewRouter(statePath, DefaultDockerSocketPath) router.RestoreLastSavedState() statusCode, body = sendGETRequest(router, "http://something.example.com/") @@ -845,7 +845,7 @@ func TestRouter_RestoreLastSavedState_TLSOnDemandURL(t *testing.T) { func testRouter(t *testing.T) *Router { statePath := filepath.Join(t.TempDir(), "state.json") - return NewRouter(statePath) + return NewRouter(statePath, DefaultDockerSocketPath) } func sendGETRequest(router *Router, url string) (int, string) { diff --git a/internal/server/service.go b/internal/server/service.go index d3c88d6d..4ddc853a 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -47,7 +47,9 @@ const ( DefaultTargetTimeout = time.Second * 30 DefaultMaxMemoryBufferSize = 1 * MB DefaultMaxRequestBodySize = 0 - DefaultMaxResponseBodySize = 0 + DefaultMaxResponseBodySize = 0 + + DefaultIdleWakeTimeout = 30 * time.Second DefaultStopMessage = "" ) @@ -111,6 +113,8 @@ type ServiceOptions struct { ReadTargetsAcceptWebsockets bool `json:"read_targets_accept_websockets"` ExcludeMetricsPaths []string `json:"exclude_metrics_paths"` ClientIPHeader string `json:"client_ip_header"` + IdleTimeout time.Duration `json:"idle_timeout"` + IdleWakeTimeout time.Duration `json:"idle_wake_timeout"` } func (so *ServiceOptions) ShouldExcludeMetrics(r *http.Request) bool { @@ -203,15 +207,19 @@ type Service struct { serviceLock sync.RWMutex pauseController *PauseController + idleController *IdleController rolloutController *RolloutController + docker *DockerClient + certManager CertManager middleware http.Handler } -func NewService(name string, options ServiceOptions, targetOptions TargetOptions) (*Service, error) { +func NewService(name string, options ServiceOptions, targetOptions TargetOptions, docker *DockerClient) (*Service, error) { service := &Service{ name: name, + docker: docker, pauseController: NewPauseController(), } @@ -226,6 +234,10 @@ func (s *Service) UpdateOptions(options ServiceOptions, targetOptions TargetOpti } func (s *Service) Dispose() { + if s.idleController != nil { + s.idleController.Close() + } + s.active.Dispose() if s.rollout != nil { s.rollout.Dispose() @@ -244,6 +256,10 @@ func (s *Service) UpdateLoadBalancer(lb *LoadBalancer, slot TargetSlot) *LoadBal } else { replaced = s.active s.active = lb + if s.idleController != nil { + s.idleController.lb = lb + s.idleController.UpdateContainers(lb.WriteTargets().Names()) + } } return replaced @@ -279,6 +295,10 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { defer metrics.Tracker.SubtractInflightRequest(s.name) } + if s.idleController != nil && !s.targetOptions.IsHealthCheckRequest(r) { + s.idleController.TrackActivity() + } + s.middleware.ServeHTTP(w, r) } @@ -291,6 +311,7 @@ type marshalledService struct { RolloutTargets []string `json:"rollout_targets"` RolloutReaders []string `json:"rollout_readers"` PauseController *PauseController `json:"pause_controller"` + IdleController *IdleController `json:"idle_controller"` RolloutController *RolloutController `json:"rollout_controller"` LegacyActiveTarget string `json:"active_target,omitempty"` @@ -316,6 +337,7 @@ func (s *Service) MarshalJSON() ([]byte, error) { Options: s.options, TargetOptions: s.targetOptions, PauseController: s.pauseController, + IdleController: s.idleController, RolloutController: s.rolloutController, }) } @@ -344,6 +366,7 @@ func (s *Service) UnmarshalJSON(data []byte) error { s.name = ms.Name s.pauseController = ms.PauseController + s.idleController = ms.IdleController s.rolloutController = ms.RolloutController activeTargets, err := NewTargetList(ms.ActiveTargets, ms.ActiveReaders, ms.TargetOptions) @@ -366,6 +389,10 @@ func (s *Service) UnmarshalJSON(data []byte) error { } func (s *Service) Stop(drainTimeout time.Duration, message string) error { + if s.idleController != nil { + s.idleController.Disable() + } + err := s.pauseController.Stop(message) if err != nil { return err @@ -379,6 +406,10 @@ func (s *Service) Stop(drainTimeout time.Duration, message string) error { } func (s *Service) Pause(drainTimeout time.Duration, pauseTimeout time.Duration) error { + if s.idleController != nil { + s.idleController.Disable() + } + err := s.pauseController.Pause(pauseTimeout) if err != nil { return err @@ -392,6 +423,10 @@ func (s *Service) Pause(drainTimeout time.Duration, pauseTimeout time.Duration) } func (s *Service) Resume() error { + if s.idleController != nil { + s.idleController.Enable() + } + err := s.pauseController.Resume() if err != nil { return err @@ -419,6 +454,22 @@ func (s *Service) initialize(options ServiceOptions, targetOptions TargetOptions s.certManager = certManager s.middleware = middleware + if s.options.IdleTimeout > 0 { + if s.idleController == nil { + s.idleController = NewIdleController(s.name, s.options.IdleTimeout, s.options.IdleWakeTimeout, s.activeTargets(), s.docker, s.active) + } else { + s.idleController.docker = s.docker + s.idleController.lb = s.active + s.idleController.serviceName = s.name + s.idleController.IdleTimeout = s.options.IdleTimeout + s.idleController.WakeTimeout = s.options.IdleWakeTimeout + s.idleController.UpdateContainers(s.activeTargets()) + } + } else if s.idleController != nil { + s.idleController.Close() + s.idleController = nil + } + return nil } @@ -445,6 +496,13 @@ func (s *Service) loadBalancerForRequest(req *http.Request) *LoadBalancer { return lb } +func (s *Service) activeTargets() []string { + if s.active == nil { + return nil + } + return s.active.WriteTargets().Names() +} + func (s *Service) servesRootPath() bool { return slices.Contains(s.options.PathPrefixes, rootPath) } @@ -535,12 +593,43 @@ func (s *Service) serviceRequestWithTarget(w http.ResponseWriter, r *http.Reques return } + if s.handleIdleRequests(w, r) { + return + } + sendRequest := s.startLoadBalancerRequest(w, r) if sendRequest != nil { sendRequest() } } +func (s *Service) handleIdleRequests(w http.ResponseWriter, r *http.Request) bool { + if s.idleController == nil { + return false + } + + if s.targetOptions.IsHealthCheckRequest(r) { + // Health checks should not wake the service. + // If it's sleeping, just return 200. + icState := s.idleController.GetState() + if icState != IdleStateActive { + w.WriteHeader(http.StatusOK) + return true + } + return false + } + + action := s.idleController.WaitIfSleeping() + if action == IdleWaitActionTimedOut { + slog.Warn("Rejecting request due to idle wake timeout", "service", s.name, "path", r.URL.Path) + w.Header().Set("Retry-After", "10") + SetErrorResponse(w, r, http.StatusServiceUnavailable, nil) + return true + } + + return false +} + func (s *Service) startLoadBalancerRequest(w http.ResponseWriter, r *http.Request) func() { s.serviceLock.RLock() defer s.serviceLock.RUnlock() diff --git a/internal/server/service_test.go b/internal/server/service_test.go index 4a74219f..8ff85dd7 100644 --- a/internal/server/service_test.go +++ b/internal/server/service_test.go @@ -364,7 +364,7 @@ func testCreateServiceWithHandler(t *testing.T, options ServiceOptions, targetOp target, err := NewTarget(serverURL.Host, targetOptions) require.NoError(t, err) - service, err := NewService("test", options, targetOptions) + service, err := NewService("test", options, targetOptions, NewDockerClient(DefaultDockerSocketPath)) require.NoError(t, err) service.UpdateLoadBalancer(NewLoadBalancer(TargetList{target}, DefaultWriterAffinityTimeout, false), TargetSlotActive) diff --git a/internal/server/testing.go b/internal/server/testing.go index 61576bf1..d58b7194 100644 --- a/internal/server/testing.go +++ b/internal/server/testing.go @@ -85,7 +85,7 @@ func testServer(t testing.TB, http3Enabled bool) *Server { AlternateConfigDir: t.TempDir(), HTTP3Enabled: http3Enabled, } - router := NewRouter(config.StatePath()) + router := NewRouter(config.StatePath(), DefaultDockerSocketPath) server := NewServer(config, router) err := server.Start() require.NoError(t, err) From 0650923bfebecbe2fc6f5287c4aee430ab4868b3 Mon Sep 17 00:00:00 2001 From: Masaki Komagata Date: Wed, 15 Jul 2026 14:27:35 +0900 Subject: [PATCH 02/17] Harden opt-in idle container lifecycle --- README.md | 26 ++ internal/cmd/util.go | 8 + internal/server/config.go | 4 +- internal/server/idle_controller.go | 373 +++++++++++++----------- internal/server/idle_controller_test.go | 130 +++++---- internal/server/load_balancer.go | 10 + internal/server/router.go | 30 +- internal/server/service.go | 79 ++--- internal/server/target.go | 8 + 9 files changed, 397 insertions(+), 271 deletions(-) diff --git a/README.md b/README.md index 2bdbb09d..eed703b9 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,32 @@ Each deployment takes over all the traffic from the previously deployed instance. As soon as Kamal Proxy determines that the new instance is healthy, it will route all new traffic to that instance. +### Opt-in scale to zero + +Services can stop their write containers after an idle period and wake them on +the next application request: + + kamal-proxy run --docker-socket /var/run/docker.sock + kamal-proxy deploy service1 --target web-1:3000 --idle-timeout 15m --idle-wake-timeout 30s + +`--idle-timeout` defaults to `0` (disabled). `--idle-wake-timeout` defaults to +`30s` and bounds how long each request waits for Docker start and a successful +configured health check. The target hostname (`web-1` above) must be the Docker +container name. `DOCKER_SOCKET` and `KAMAL_PROXY_DOCKER_SOCKET` are equivalents +of the run flag. + +Requests are held before their bodies are read, so POST bodies are forwarded +unchanged after a successful wake. Concurrent wake requests are coalesced. +Open streaming responses and WebSockets count as activity/in-flight work and +prevent sleeping until they close; a new stream or WebSocket is held during +wake like any other request. Health-check requests do not wake or reset an idle +service and receive success while it is sleeping. + +Mounting the Docker socket gives the proxy host-level container control. Only +enable this feature where that trust is acceptable; the lifecycle calls are +isolated behind the `ContainerLifecycle` interface so they can be moved to an +external service later. + The `deploy` command also waits for traffic to drain from the old instance before returning. This means it's safe to remove the old instance as soon as `deploy` returns successfully, without interrupting any in-flight requests. diff --git a/internal/cmd/util.go b/internal/cmd/util.go index d28618c6..000a303b 100644 --- a/internal/cmd/util.go +++ b/internal/cmd/util.go @@ -47,6 +47,14 @@ func getEnvInt(key string, defaultValue int) int { return intValue } +func getEnvString(key, defaultValue string) string { + value, ok := findEnv(key) + if !ok { + return defaultValue + } + return value +} + func getEnvBool(key string, defaultValue bool) bool { value, ok := findEnv(key) if !ok { diff --git a/internal/server/config.go b/internal/server/config.go index 052b069a..92621f06 100644 --- a/internal/server/config.go +++ b/internal/server/config.go @@ -8,8 +8,8 @@ import ( ) const ( - DefaultHttpPort = 80 - DefaultHttpsPort = 443 + DefaultHttpPort = 80 + DefaultHttpsPort = 443 DefaultDockerSocketPath = "/var/run/docker.sock" ) diff --git a/internal/server/idle_controller.go b/internal/server/idle_controller.go index 7486aa6a..fc98acfa 100644 --- a/internal/server/idle_controller.go +++ b/internal/server/idle_controller.go @@ -3,6 +3,7 @@ package server import ( "context" "encoding/json" + "errors" "log/slog" "sync" "time" @@ -12,14 +13,17 @@ type IdleState int const ( IdleStateActive IdleState = iota + IdleStateStopping IdleStateSleeping IdleStateWaking ) -func (is IdleState) String() string { - switch is { +func (s IdleState) String() string { + switch s { case IdleStateActive: return "active" + case IdleStateStopping: + return "stopping" case IdleStateSleeping: return "sleeping" case IdleStateWaking: @@ -29,12 +33,12 @@ func (is IdleState) String() string { } } -type IdleWaitAction int +var ErrIdleWakeTimeout = errors.New("idle container wake timed out") -const ( - IdleWaitActionProceed IdleWaitAction = iota - IdleWaitActionTimedOut -) +type ContainerLifecycle interface { + StartContainer(context.Context, string) error + StopContainer(context.Context, string) error +} type IdleController struct { State IdleState `json:"state"` @@ -42,211 +46,242 @@ type IdleController struct { WakeTimeout time.Duration `json:"wake_timeout"` ContainerNames []string `json:"container_names"` - serviceName string - docker *DockerClient - lb *LoadBalancer - - lock sync.RWMutex - lastRequestAt time.Time - wakeChan chan bool - closeChan chan bool - disabled bool -} - -func NewIdleController(serviceName string, idleTimeout, wakeTimeout time.Duration, containerNames []string, docker *DockerClient, lb *LoadBalancer) *IdleController { - ic := &IdleController{ - State: IdleStateActive, - IdleTimeout: idleTimeout, - WakeTimeout: wakeTimeout, - ContainerNames: containerNames, - serviceName: serviceName, - docker: docker, - lb: lb, - lastRequestAt: time.Now(), - closeChan: make(chan bool), - } + mu sync.Mutex + lifecycle ContainerLifecycle + ready func(time.Duration) error + inflight int + lastRequest time.Time + wakeDone chan struct{} + wakeErr error + changed chan struct{} + closed chan struct{} + disabled bool + closeOnce sync.Once + persist func() +} - go ic.run() - return ic +func NewIdleController(idleTimeout, wakeTimeout time.Duration, names []string, lifecycle ContainerLifecycle, ready func(time.Duration) error) *IdleController { + c := &IdleController{State: IdleStateActive} + c.configure(idleTimeout, wakeTimeout, names, lifecycle, ready) + return c } -func (ic *IdleController) UnmarshalJSON(data []byte) error { - type alias IdleController - aux := &struct { - *alias - }{ - alias: (*alias)(ic), +func (c *IdleController) UnmarshalJSON(data []byte) error { + type persisted struct { + State IdleState `json:"state"` + IdleTimeout time.Duration `json:"idle_timeout"` + WakeTimeout time.Duration `json:"wake_timeout"` + ContainerNames []string `json:"container_names"` } - if err := json.Unmarshal(data, &aux); err != nil { + var p persisted + if err := json.Unmarshal(data, &p); err != nil { return err } - - ic.lastRequestAt = time.Now() - ic.closeChan = make(chan bool) - - go ic.run() + c.State, c.IdleTimeout, c.WakeTimeout, c.ContainerNames = p.State, p.IdleTimeout, p.WakeTimeout, p.ContainerNames + if c.State == IdleStateWaking || c.State == IdleStateStopping { + c.State = IdleStateSleeping + } return nil } -func (ic *IdleController) TrackActivity() { - ic.lock.Lock() - defer ic.lock.Unlock() - - ic.lastRequestAt = time.Now() +func (c *IdleController) configure(idleTimeout, wakeTimeout time.Duration, names []string, lifecycle ContainerLifecycle, ready func(time.Duration) error) { + c.mu.Lock() + defer c.mu.Unlock() + c.IdleTimeout, c.WakeTimeout = idleTimeout, wakeTimeout + c.ContainerNames = append([]string(nil), names...) + c.lifecycle, c.ready = lifecycle, ready + c.lastRequest = time.Now() + if c.changed == nil { + c.changed, c.closed = make(chan struct{}, 1), make(chan struct{}) + go c.run() + } + c.signal() } -func (ic *IdleController) GetState() IdleState { - ic.lock.RLock() - defer ic.lock.RUnlock() - return ic.State +func (c *IdleController) BeginRequest(ctx context.Context) error { + c.mu.Lock() + c.inflight++ + c.lastRequest = time.Now() + c.mu.Unlock() + c.signal() + for { + c.mu.Lock() + if c.State == IdleStateSleeping && !c.disabled { + c.startWakeLocked() + } + done, timeout, state := c.wakeDone, c.WakeTimeout, c.State + c.mu.Unlock() + if state != IdleStateWaking && state != IdleStateStopping { + return nil + } + timer := time.NewTimer(timeout) + select { + case <-done: + timer.Stop() + if state == IdleStateStopping { + continue + } + c.mu.Lock() + err := c.wakeErr + c.mu.Unlock() + return err + case <-timer.C: + return ErrIdleWakeTimeout + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + } + } } -func (ic *IdleController) WaitIfSleeping() IdleWaitAction { - ic.lock.RLock() - state := ic.State - wakeChan := ic.wakeChan - wakeTimeout := ic.WakeTimeout - ic.lock.RUnlock() +func (c *IdleController) EndRequest() { + c.mu.Lock() + c.inflight-- + c.lastRequest = time.Now() + c.mu.Unlock() + c.signal() +} - if state == IdleStateActive { - return IdleWaitActionProceed - } +func (c *IdleController) StateValue() IdleState { c.mu.Lock(); defer c.mu.Unlock(); return c.State } - if state == IdleStateSleeping { - ic.wake() - // Re-read wakeChan - ic.lock.RLock() - wakeChan = ic.wakeChan - ic.lock.RUnlock() - } +func (c *IdleController) SetPersist(fn func()) { c.mu.Lock(); c.persist = fn; c.mu.Unlock() } - if wakeChan == nil { - return IdleWaitActionProceed - } - - select { - case <-wakeChan: - return IdleWaitActionProceed - case <-time.After(wakeTimeout): - return IdleWaitActionTimedOut +func (c *IdleController) notifyPersist() { + c.mu.Lock() + fn := c.persist + c.mu.Unlock() + if fn != nil { + go fn() } } -func (ic *IdleController) UpdateContainers(names []string) { - ic.lock.Lock() - defer ic.lock.Unlock() +func (c *IdleController) Disable() { c.mu.Lock(); c.disabled = true; c.mu.Unlock(); c.signal() } +func (c *IdleController) Enable() { + c.mu.Lock() + c.disabled = false + c.lastRequest = time.Now() + c.mu.Unlock() + c.signal() +} - ic.ContainerNames = names - ic.lastRequestAt = time.Now() - - if ic.State != IdleStateActive { - ic.State = IdleStateActive - if ic.wakeChan != nil { - close(ic.wakeChan) - ic.wakeChan = nil +func (c *IdleController) Reset(names []string, ready func(time.Duration) error) { + c.mu.Lock() + c.ContainerNames = append([]string(nil), names...) + c.ready = ready + c.State, c.wakeErr, c.lastRequest = IdleStateActive, nil, time.Now() + if c.wakeDone != nil { + select { + case <-c.wakeDone: + default: + close(c.wakeDone) } + c.wakeDone = nil } + c.mu.Unlock() + c.signal() } -func (ic *IdleController) Disable() { - ic.lock.Lock() - defer ic.lock.Unlock() - ic.disabled = true -} - -func (ic *IdleController) Enable() { - ic.lock.Lock() - defer ic.lock.Unlock() - ic.disabled = false - ic.lastRequestAt = time.Now() +func (c *IdleController) Close() { + c.closeOnce.Do(func() { + if c.closed != nil { + close(c.closed) + } + }) } - -func (ic *IdleController) Close() { - close(ic.closeChan) +func (c *IdleController) signal() { + select { + case c.changed <- struct{}{}: + default: + } } -func (ic *IdleController) run() { - ticker := time.NewTicker(10 * time.Second) - defer ticker.Stop() - +func (c *IdleController) run() { for { + c.mu.Lock() + wait := c.IdleTimeout - time.Since(c.lastRequest) + eligible := !c.disabled && c.State == IdleStateActive && c.inflight == 0 && c.IdleTimeout > 0 + c.mu.Unlock() + if !eligible { + wait = time.Hour + } + if wait < 0 { + wait = 0 + } + timer := time.NewTimer(wait) select { - case <-ic.closeChan: + case <-timer.C: + c.trySleep() + case <-c.changed: + if !timer.Stop() { + <-timer.C + } + case <-c.closed: + if !timer.Stop() { + <-timer.C + } return - case <-ticker.C: - ic.checkIdle() } } } -func (ic *IdleController) checkIdle() { - ic.lock.Lock() - if ic.disabled || ic.State != IdleStateActive || ic.IdleTimeout <= 0 { - ic.lock.Unlock() +func (c *IdleController) trySleep() { + c.mu.Lock() + if c.disabled || c.State != IdleStateActive || c.inflight != 0 || c.IdleTimeout <= 0 || time.Since(c.lastRequest) < c.IdleTimeout { + c.mu.Unlock() return } - - if time.Since(ic.lastRequestAt) > ic.IdleTimeout { - ic.State = IdleStateSleeping - ic.wakeChan = make(chan bool) - containerNames := ic.ContainerNames - ic.lock.Unlock() - - slog.Info("Service is idle, stopping containers", "service", ic.serviceName, "containers", containerNames) - - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - - for _, name := range containerNames { - if err := ic.docker.StopContainer(ctx, name); err != nil { - slog.Error("Failed to stop idle container", "service", ic.serviceName, "container", name, "error", err) - } + names, lifecycle := append([]string(nil), c.ContainerNames...), c.lifecycle + c.State, c.wakeDone = IdleStateStopping, make(chan struct{}) + stopDone := c.wakeDone + c.mu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), DefaultIdleLifecycleTimeout) + defer cancel() + for _, name := range names { + if err := lifecycle.StopContainer(ctx, name); err != nil { + slog.Error("Failed to stop idle container", "container", name, "error", err) + c.mu.Lock() + c.State, c.lastRequest = IdleStateActive, time.Now() + close(stopDone) + c.mu.Unlock() + c.notifyPersist() + return } - } else { - ic.lock.Unlock() } + c.mu.Lock() + c.State = IdleStateSleeping + close(stopDone) + c.mu.Unlock() + c.notifyPersist() + c.signal() } -func (ic *IdleController) wake() { - ic.lock.Lock() - if ic.State != IdleStateSleeping { - ic.lock.Unlock() - return - } - - ic.State = IdleStateWaking - containerNames := ic.ContainerNames - ic.lock.Unlock() - - slog.Info("Service waking up, starting containers", "service", ic.serviceName, "containers", containerNames) - +func (c *IdleController) startWakeLocked() { + c.State, c.wakeErr, c.wakeDone = IdleStateWaking, nil, make(chan struct{}) + done, names, timeout, lifecycle, ready := c.wakeDone, append([]string(nil), c.ContainerNames...), c.WakeTimeout, c.lifecycle, c.ready go func() { - ctx, cancel := context.WithTimeout(context.Background(), ic.WakeTimeout) + ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - - var wg sync.WaitGroup - wg.Add(len(containerNames)) - for _, name := range containerNames { - go func(n string) { - defer wg.Done() - if err := ic.docker.StartContainer(ctx, n); err != nil { - slog.Error("Failed to start container during wake", "service", ic.serviceName, "container", n, "error", err) - } - }(name) + var err error + for _, name := range names { + if err = lifecycle.StartContainer(ctx, name); err != nil { + break + } } - wg.Wait() - - // Wait until healthy - err := ic.lb.WaitUntilHealthy(ic.WakeTimeout) - if err != nil { - slog.Error("Service failed to become healthy after wake", "service", ic.serviceName, "error", err) + if err == nil { + err = ready(timeout) } - - ic.lock.Lock() - ic.State = IdleStateActive - ic.lastRequestAt = time.Now() - close(ic.wakeChan) - ic.wakeChan = nil - ic.lock.Unlock() + c.mu.Lock() + if c.wakeDone == done { + c.wakeErr = err + if err == nil { + c.State, c.lastRequest = IdleStateActive, time.Now() + } else { + c.State = IdleStateSleeping + } + close(done) + } + c.mu.Unlock() + c.notifyPersist() + c.signal() }() } diff --git a/internal/server/idle_controller_test.go b/internal/server/idle_controller_test.go index 11dc5215..ab213711 100644 --- a/internal/server/idle_controller_test.go +++ b/internal/server/idle_controller_test.go @@ -2,9 +2,10 @@ package server import ( "context" - "net" - "net/http" - "net/http/httptest" + "encoding/json" + "errors" + "sync" + "sync/atomic" "testing" "time" @@ -12,64 +13,83 @@ import ( "github.com/stretchr/testify/require" ) -func TestIdleController_IdleAndWake(t *testing.T) { - stopCalled := make(chan bool, 1) - startCalled := make(chan bool, 1) - - dockerServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path == "/v1.41/containers/test-container/stop" { - stopCalled <- true - w.WriteHeader(http.StatusNoContent) - } else if r.URL.Path == "/v1.41/containers/test-container/start" { - startCalled <- true - w.WriteHeader(http.StatusNoContent) - } - })) - - socketPath := t.TempDir() + "/docker.sock" - l, err := net.Listen("unix", socketPath) - require.NoError(t, err) - dockerServer.Listener = l - dockerServer.Start() - defer dockerServer.Close() - - dockerClient := NewDockerClient(socketPath) - - backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - w.Write([]byte("ok")) - })) - defer backend.Close() +type fakeLifecycle struct { + starts, stops atomic.Int32 + startErr, stopErr error +} - target, _ := NewTarget(backend.URL[7:], defaultTargetOptions) - tl := TargetList{target} - lb := NewLoadBalancer(tl, 0, false) - lb.MarkAllHealthy() +func (f *fakeLifecycle) StartContainer(context.Context, string) error { + f.starts.Add(1) + return f.startErr +} +func (f *fakeLifecycle) StopContainer(context.Context, string) error { + f.stops.Add(1) + return f.stopErr +} - ic := NewIdleController("test", 100*time.Millisecond, time.Second, []string{"test-container"}, dockerClient, lb) - defer ic.Close() +func waitFor(t *testing.T, check func() bool) { + t.Helper() + require.Eventually(t, check, time.Second, time.Millisecond) +} - // Initial state: active - assert.Equal(t, IdleStateActive, ic.GetState()) +func TestIdleControllerDoesNotStopInflightRequest(t *testing.T) { + lifecycle := &fakeLifecycle{} + c := NewIdleController(10*time.Millisecond, time.Second, []string{"web"}, lifecycle, func(time.Duration) error { return nil }) + defer c.Close() + require.NoError(t, c.BeginRequest(context.Background())) + time.Sleep(30 * time.Millisecond) + assert.Zero(t, lifecycle.stops.Load()) + c.EndRequest() + waitFor(t, func() bool { return lifecycle.stops.Load() == 1 }) +} - // Wait for idle to trigger - select { - case <-stopCalled: - case <-time.After(time.Second): - t.Fatal("Timeout waiting for container to be stopped") +func TestIdleControllerCoalescesConcurrentWake(t *testing.T) { + lifecycle := &fakeLifecycle{} + ready := make(chan struct{}) + c := NewIdleController(time.Millisecond, time.Second, []string{"web"}, lifecycle, func(time.Duration) error { <-ready; return nil }) + defer c.Close() + waitFor(t, func() bool { return c.StateValue() == IdleStateSleeping }) + var wg sync.WaitGroup + for range 10 { + wg.Add(1) + go func() { defer wg.Done(); require.NoError(t, c.BeginRequest(context.Background())); c.EndRequest() }() } + waitFor(t, func() bool { return lifecycle.starts.Load() == 1 }) + close(ready) + wg.Wait() + assert.Equal(t, int32(1), lifecycle.starts.Load()) +} - assert.Equal(t, IdleStateSleeping, ic.GetState()) - - // Wake up on request - action := ic.WaitIfSleeping() - assert.Equal(t, IdleWaitActionProceed, action) +func TestIdleControllerWakeFailureAndTimeout(t *testing.T) { + lifecycle := &fakeLifecycle{startErr: errors.New("start failed")} + c := NewIdleController(time.Millisecond, time.Second, []string{"web"}, lifecycle, func(time.Duration) error { return nil }) + defer c.Close() + waitFor(t, func() bool { return c.StateValue() == IdleStateSleeping }) + assert.ErrorContains(t, c.BeginRequest(context.Background()), "start failed") + c.EndRequest() - select { - case <-startCalled: - case <-time.After(time.Second): - t.Fatal("Timeout waiting for container to be started") - } + blocking := make(chan struct{}) + c.Reset([]string{"web"}, func(time.Duration) error { <-blocking; return nil }) + c.mu.Lock() + c.State, c.WakeTimeout = IdleStateSleeping, 10*time.Millisecond + c.mu.Unlock() + lifecycle.startErr = nil + assert.ErrorIs(t, c.BeginRequest(context.Background()), ErrIdleWakeTimeout) + c.EndRequest() + close(blocking) +} - assert.Equal(t, IdleStateActive, ic.GetState()) +func TestIdleControllerRestoresSleepingAndWakes(t *testing.T) { + original := &IdleController{State: IdleStateWaking, IdleTimeout: time.Minute, WakeTimeout: time.Second, ContainerNames: []string{"web"}} + data, err := json.Marshal(original) + require.NoError(t, err) + var restored IdleController + require.NoError(t, json.Unmarshal(data, &restored)) + lifecycle := &fakeLifecycle{} + restored.configure(time.Minute, time.Second, []string{"web"}, lifecycle, func(time.Duration) error { return nil }) + defer restored.Close() + assert.Equal(t, IdleStateSleeping, restored.StateValue()) + require.NoError(t, restored.BeginRequest(context.Background())) + restored.EndRequest() + assert.Equal(t, int32(1), lifecycle.starts.Load()) } diff --git a/internal/server/load_balancer.go b/internal/server/load_balancer.go index 6bec9aec..1c494ef6 100644 --- a/internal/server/load_balancer.go +++ b/internal/server/load_balancer.go @@ -153,6 +153,16 @@ func (lb *LoadBalancer) MarkAllHealthy() { lb.updateHealthyTargets() } +func (lb *LoadBalancer) PrepareForWake() { + for _, target := range lb.all { + target.markUnhealthyForWake() + } + lb.lock.Lock() + lb.waitForHealthyContext, lb.markHealthy = context.WithCancel(context.Background()) + lb.writers, lb.readers = TargetList{}, TargetList{} + lb.lock.Unlock() +} + func (lb *LoadBalancer) Dispose() { lb.all.StopHealthChecks() } diff --git a/internal/server/router.go b/internal/server/router.go index 792e8df8..15bf805c 100644 --- a/internal/server/router.go +++ b/internal/server/router.go @@ -51,6 +51,7 @@ type Router struct { dockerClient *DockerClient services *ServiceMap serviceLock sync.RWMutex + stateLock sync.Mutex } type ServiceDescription struct { @@ -90,19 +91,19 @@ func (r *Router) RestoreLastSavedState() error { return err } - r.withWriteLock(func() error { + return r.withWriteLock(func() error { r.services = NewServiceMap() for _, service := range services { - service.docker = r.dockerClient - service.initialize(service.options, service.targetOptions) + service.lifecycle = r.dockerClient + service.stateChanged = func() { _ = r.saveStateSnapshot() } + if err := service.initialize(service.options, service.targetOptions); err != nil { + return err + } r.services.Set(service) } - + slog.Info("Restored saved state", "path", r.statePath) return nil }) - - slog.Info("Restored saved state", "path", r.statePath) - return nil } func (r *Router) ServeHTTP(w http.ResponseWriter, req *http.Request) { @@ -265,7 +266,7 @@ func (r *Router) ListActiveServices() ServiceDescriptionMap { state := service.pauseController.GetState().String() if service.idleController != nil && service.pauseController.GetState() == PauseStateRunning { - icState := service.idleController.GetState() + icState := service.idleController.StateValue() if icState != IdleStateActive { state = icState.String() } @@ -317,10 +318,17 @@ func (r *Router) GetCertificate(hello *tls.ClientHelloInfo) (*tls.Certificate, e func (r *Router) createOrUpdateService(name string, options ServiceOptions, targetOptions TargetOptions) (*Service, error) { service := r.services.Get(name) if service == nil { - return NewService(name, options, targetOptions, r.dockerClient) + service, err := NewService(name, options, targetOptions, r.dockerClient) + if err == nil { + service.stateChanged = func() { _ = r.saveStateSnapshot() } + if service.idleController != nil { + service.idleController.SetPersist(service.stateChanged) + } + } + return service, err } - service.docker = r.dockerClient + service.lifecycle = r.dockerClient err := service.UpdateOptions(options, targetOptions) return service, err } @@ -370,6 +378,8 @@ func (r *Router) installLoadBalancer(name string, slot TargetSlot, lb *LoadBalan } func (r *Router) saveStateSnapshot() error { + r.stateLock.Lock() + defer r.stateLock.Unlock() services := []*Service{} r.withReadLock(func() error { for _, service := range r.services.All() { diff --git a/internal/server/service.go b/internal/server/service.go index 4ddc853a..32334e2e 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -47,9 +47,10 @@ const ( DefaultTargetTimeout = time.Second * 30 DefaultMaxMemoryBufferSize = 1 * MB DefaultMaxRequestBodySize = 0 - DefaultMaxResponseBodySize = 0 + DefaultMaxResponseBodySize = 0 - DefaultIdleWakeTimeout = 30 * time.Second + DefaultIdleWakeTimeout = 30 * time.Second + DefaultIdleLifecycleTimeout = 30 * time.Second DefaultStopMessage = "" ) @@ -124,6 +125,9 @@ func (so *ServiceOptions) ShouldExcludeMetrics(r *http.Request) bool { func (so *ServiceOptions) Normalize() { so.Hosts = NormalizeHosts(so.Hosts) so.PathPrefixes = NormalizePathPrefixes(so.PathPrefixes) + if so.IdleTimeout > 0 && so.IdleWakeTimeout == 0 { + so.IdleWakeTimeout = DefaultIdleWakeTimeout + } } func (so ServiceOptions) Validate() error { @@ -164,6 +168,9 @@ func (so ServiceOptions) Validate() error { return fmt.Errorf("%w: canonical-host '%s' must be present in the hosts list: %v", ErrServiceOptionsInvalid, so.CanonicalHost, so.Hosts) } } + if so.IdleTimeout < 0 || so.IdleWakeTimeout < 0 { + return fmt.Errorf("%w: idle timeouts cannot be negative", ErrServiceOptionsInvalid) + } return nil } @@ -210,16 +217,17 @@ type Service struct { idleController *IdleController rolloutController *RolloutController - docker *DockerClient + lifecycle ContainerLifecycle - certManager CertManager - middleware http.Handler + certManager CertManager + middleware http.Handler + stateChanged func() } -func NewService(name string, options ServiceOptions, targetOptions TargetOptions, docker *DockerClient) (*Service, error) { +func NewService(name string, options ServiceOptions, targetOptions TargetOptions, lifecycle ContainerLifecycle) (*Service, error) { service := &Service{ name: name, - docker: docker, + lifecycle: lifecycle, pauseController: NewPauseController(), } @@ -257,8 +265,7 @@ func (s *Service) UpdateLoadBalancer(lb *LoadBalancer, slot TargetSlot) *LoadBal replaced = s.active s.active = lb if s.idleController != nil { - s.idleController.lb = lb - s.idleController.UpdateContainers(lb.WriteTargets().Names()) + s.idleController.Reset(s.activeContainerNames(), s.waitUntilActiveHealthy) } } @@ -296,7 +303,12 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { } if s.idleController != nil && !s.targetOptions.IsHealthCheckRequest(r) { - s.idleController.TrackActivity() + if err := s.idleController.BeginRequest(r.Context()); err != nil { + s.idleController.EndRequest() + SetErrorResponse(w, r, http.StatusServiceUnavailable, err) + return + } + defer s.idleController.EndRequest() } s.middleware.ServeHTTP(w, r) @@ -456,15 +468,11 @@ func (s *Service) initialize(options ServiceOptions, targetOptions TargetOptions if s.options.IdleTimeout > 0 { if s.idleController == nil { - s.idleController = NewIdleController(s.name, s.options.IdleTimeout, s.options.IdleWakeTimeout, s.activeTargets(), s.docker, s.active) + s.idleController = NewIdleController(s.options.IdleTimeout, s.options.IdleWakeTimeout, s.activeContainerNames(), s.lifecycle, s.waitUntilActiveHealthy) } else { - s.idleController.docker = s.docker - s.idleController.lb = s.active - s.idleController.serviceName = s.name - s.idleController.IdleTimeout = s.options.IdleTimeout - s.idleController.WakeTimeout = s.options.IdleWakeTimeout - s.idleController.UpdateContainers(s.activeTargets()) + s.idleController.configure(s.options.IdleTimeout, s.options.IdleWakeTimeout, s.activeContainerNames(), s.lifecycle, s.waitUntilActiveHealthy) } + s.idleController.SetPersist(s.stateChanged) } else if s.idleController != nil { s.idleController.Close() s.idleController = nil @@ -473,6 +481,22 @@ func (s *Service) initialize(options ServiceOptions, targetOptions TargetOptions return nil } +func (s *Service) activeContainerNames() []string { + names := make([]string, 0, len(s.active.WriteTargets())) + for _, target := range s.active.WriteTargets() { + names = append(names, target.ContainerName()) + } + return names +} + +func (s *Service) waitUntilActiveHealthy(timeout time.Duration) error { + s.serviceLock.RLock() + lb := s.active + s.serviceLock.RUnlock() + lb.PrepareForWake() + return lb.WaitUntilHealthy(timeout) +} + func (s *Service) Drain(timeout time.Duration) { PerformConcurrently( func() { @@ -496,13 +520,6 @@ func (s *Service) loadBalancerForRequest(req *http.Request) *LoadBalancer { return lb } -func (s *Service) activeTargets() []string { - if s.active == nil { - return nil - } - return s.active.WriteTargets().Names() -} - func (s *Service) servesRootPath() bool { return slices.Contains(s.options.PathPrefixes, rootPath) } @@ -593,7 +610,7 @@ func (s *Service) serviceRequestWithTarget(w http.ResponseWriter, r *http.Reques return } - if s.handleIdleRequests(w, r) { + if s.handleIdleHealthCheck(w, r) { return } @@ -603,7 +620,7 @@ func (s *Service) serviceRequestWithTarget(w http.ResponseWriter, r *http.Reques } } -func (s *Service) handleIdleRequests(w http.ResponseWriter, r *http.Request) bool { +func (s *Service) handleIdleHealthCheck(w http.ResponseWriter, r *http.Request) bool { if s.idleController == nil { return false } @@ -611,7 +628,7 @@ func (s *Service) handleIdleRequests(w http.ResponseWriter, r *http.Request) boo if s.targetOptions.IsHealthCheckRequest(r) { // Health checks should not wake the service. // If it's sleeping, just return 200. - icState := s.idleController.GetState() + icState := s.idleController.StateValue() if icState != IdleStateActive { w.WriteHeader(http.StatusOK) return true @@ -619,14 +636,6 @@ func (s *Service) handleIdleRequests(w http.ResponseWriter, r *http.Request) boo return false } - action := s.idleController.WaitIfSleeping() - if action == IdleWaitActionTimedOut { - slog.Warn("Rejecting request due to idle wake timeout", "service", s.name, "path", r.URL.Path) - w.Header().Set("Retry-After", "10") - SetErrorResponse(w, r, http.StatusServiceUnavailable, nil) - return true - } - return false } diff --git a/internal/server/target.go b/internal/server/target.go index b1bd627b..4faac43e 100644 --- a/internal/server/target.go +++ b/internal/server/target.go @@ -143,6 +143,14 @@ func (t *Target) Address() string { return t.targetURL.Host } +func (t *Target) ContainerName() string { return t.targetURL.Hostname() } + +func (t *Target) markUnhealthyForWake() { + t.inflightLock.Lock() + t.state = TargetStateUnhealthy + t.inflightLock.Unlock() +} + func (t *Target) State() TargetState { t.inflightLock.Lock() defer t.inflightLock.Unlock() From e5d6a7ca7d1ec659671393c2d38ce5276b24acef Mon Sep 17 00:00:00 2001 From: Masaki Komagata Date: Wed, 15 Jul 2026 14:28:21 +0900 Subject: [PATCH 03/17] Document idle controller verification --- idle-controller-report.md | 82 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 idle-controller-report.md diff --git a/idle-controller-report.md b/idle-controller-report.md new file mode 100644 index 00000000..08c7f2ad --- /dev/null +++ b/idle-controller-report.md @@ -0,0 +1,82 @@ +# Proxy idle controller implementation report + +## Assumptions and success criteria + +- Scale-to-zero is explicitly enabled per service only when `--idle-timeout` is greater than zero. +- Each write target hostname is also its Docker container name; ports are stripped before lifecycle calls. +- The proxy process can access the configured Docker Unix socket when the feature is enabled. +- Success means: never stop with an application request in flight; serialize stop versus arriving requests; coalesce concurrent wakes; hold requests until readiness, failure, cancellation, or timeout; preserve sleeping state over restart; and keep timeout-zero behavior unchanged. +- Health-check requests neither reset idle time nor wake a sleeping service. Paused/stopped services disable idle transitions; resume restarts the idle timer; deploy replaces the lifecycle target set and returns the controller to active. + +## CLI contract + +```text +kamal-proxy run [--docker-socket PATH] + --docker-socket default: /var/run/docker.sock + environment: DOCKER_SOCKET or KAMAL_PROXY_DOCKER_SOCKET + +kamal-proxy deploy SERVICE --target CONTAINER:PORT \ + [--idle-timeout DURATION] [--idle-wake-timeout DURATION] + --idle-timeout default: 0 (disabled) + --idle-wake-timeout default: 30s when idle is enabled + negative idle durations: rejected +``` + +The request is held before its body is read, so POST bodies are unchanged. WebSockets and streaming responses remain in flight until their handler returns and therefore prevent sleep; a newly arriving WebSocket/stream waits through wake like an ordinary request. + +## Implementation + +- `ContainerLifecycle` is the narrow start/stop interface. `DockerClient` is its Unix-socket implementation and can later be replaced by an external lifecycle client without changing the controller. +- The controller uses active/stopping/sleeping/waking transitions. A stopping barrier closes the race between the zero-inflight check and Docker stop; concurrent arrivals then share a single wake result. +- Start success is not enough: held requests wait for the active load balancer to observe a healthy target. Start, readiness, cancellation, and timeout errors release all waiters without consuming request bodies. +- Idle state is serialized in the existing state file. Incomplete stopping/waking transitions recover as sleeping and wake on the next application request. State transitions trigger serialized snapshots. + +## Changed files + +- `README.md` +- `internal/cmd/deploy.go` +- `internal/cmd/run.go` +- `internal/cmd/util.go` +- `internal/server/config.go` +- `internal/server/docker_client.go` +- `internal/server/docker_client_test.go` +- `internal/server/idle_controller.go` +- `internal/server/idle_controller_test.go` +- `internal/server/load_balancer.go` +- `internal/server/router.go` +- `internal/server/router_test.go` +- `internal/server/service.go` +- `internal/server/service_test.go` +- `internal/server/target.go` +- `internal/server/testing.go` + +Local commits: `105c829` (adapted PR #197) and `5d838ea` (race/failure/restart hardening and documentation). Nothing was pushed and no PR was opened. + +## Verification + +- `go test ./...` — pass +- `go test -race ./internal/server ./internal/cmd` — pass +- `go vet ./...` — pass +- `git diff --check` — pass +- `golangci-lint run` — not run; executable is not installed (`command not found`) + +Tests cover no-stop-with-inflight, stop completion versus arriving requests, concurrent wake coalescing, start failure, wake timeout, and persisted incomplete-state recovery. + +## Unresolved security and maintainer decisions + +- Mounting Docker's socket grants powerful host/container control. Maintainers must decide whether direct socket access is acceptable, whether a restricted socket proxy/external lifecycle service is required, and how container names should be authorized per service. +- The minimal client pins Docker API `v1.41`; maintainers should decide whether API negotiation is required for supported Docker versions. +- Only write targets are stopped. Reader targets and rollout targets remain running; expanding lifecycle scope needs an explicit product decision. +- Multi-container stop is sequential and not transactional. A later stop failure can leave earlier containers stopped; the controller returns active and logs the failure. Desired rollback/reconciliation semantics need a maintainer decision. +- Persisted sleeping state trusts that Docker state still matches the snapshot. The next request issues idempotent starts, but there is no startup reconciliation/list permission. + +## Recommended real-Docker integration scenarios + +1. Single container: idle stop, POST wake with a large/chunked body, health readiness, exact response verification. +2. Twenty concurrent GET/POST requests against one sleeping service: assert one Docker start and all bodies/results preserved. +3. Long SSE response and WebSocket: assert no stop while connected, then stop after close plus idle duration. +4. Restart kamal-proxy while sleeping and during a forced stopping/waking interruption; assert the next request recovers. +5. Pause/stop/resume and deploy while active, sleeping, and waking; verify no stale container is started and the new target set wins. +6. Docker socket unavailable, permission denied, start/stop HTTP errors, slow Docker response, and container missing/already started/stopped. +7. Container starts but health never succeeds, succeeds near the deadline, or flaps; verify waiter status and retry behavior. +8. Multiple write targets with a partial stop/start failure to decide and validate reconciliation semantics. From e44dacb7e5abfdaf88f8f5a5f0f956eb461f4efb Mon Sep 17 00:00:00 2001 From: komagata Date: Wed, 15 Jul 2026 21:58:04 +0900 Subject: [PATCH 04/17] Fix idle setup before first target install --- internal/server/service.go | 3 +++ internal/server/service_test.go | 11 +++++++++++ 2 files changed, 14 insertions(+) diff --git a/internal/server/service.go b/internal/server/service.go index 32334e2e..21aca9eb 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -482,6 +482,9 @@ func (s *Service) initialize(options ServiceOptions, targetOptions TargetOptions } func (s *Service) activeContainerNames() []string { + if s.active == nil { + return nil + } names := make([]string, 0, len(s.active.WriteTargets())) for _, target := range s.active.WriteTargets() { names = append(names, target.ContainerName()) diff --git a/internal/server/service_test.go b/internal/server/service_test.go index 8ff85dd7..e7694c0b 100644 --- a/internal/server/service_test.go +++ b/internal/server/service_test.go @@ -347,6 +347,17 @@ func TestService_UnmarshallingStateFromLegacyFormat(t *testing.T) { assert.Equal(t, 3*time.Second, service.targetOptions.ResponseTimeout) } +func TestNewServiceWithIdleLifecycleBeforeActiveLoadBalancer(t *testing.T) { + service, err := NewService("test", ServiceOptions{ + IdleTimeout: time.Minute, + IdleWakeTimeout: time.Second, + }, defaultTargetOptions, &fakeLifecycle{}) + require.NoError(t, err) + t.Cleanup(service.idleController.Close) + + assert.Empty(t, service.idleController.ContainerNames) +} + func testCreateService(t *testing.T, options ServiceOptions, targetOptions TargetOptions) *Service { return testCreateServiceWithHandler( t, options, targetOptions, From 614440784c7c6a0b787fc74a4ebfdcad67823af2 Mon Sep 17 00:00:00 2001 From: komagata Date: Thu, 16 Jul 2026 01:41:38 +0900 Subject: [PATCH 05/17] Restart health checks when waking idle targets --- internal/server/health_check.go | 13 +++++++++++- internal/server/load_balancer.go | 7 ++++--- internal/server/load_balancer_test.go | 29 +++++++++++++++++++++++++++ internal/server/target.go | 9 ++++++++- 4 files changed, 53 insertions(+), 5 deletions(-) diff --git a/internal/server/health_check.go b/internal/server/health_check.go index 59eeba30..f8d48ead 100644 --- a/internal/server/health_check.go +++ b/internal/server/health_check.go @@ -8,6 +8,7 @@ import ( "log/slog" "net/http" "net/url" + "sync" "time" ) @@ -33,9 +34,16 @@ type HealthCheck struct { ctx context.Context cancel context.CancelFunc + start sync.Once } func NewHealthCheck(consumer HealthCheckConsumer, endpoint *url.URL, interval time.Duration, timeout time.Duration, host string) *HealthCheck { + hc := newHealthCheck(consumer, endpoint, interval, timeout, host) + hc.Start() + return hc +} + +func newHealthCheck(consumer HealthCheckConsumer, endpoint *url.URL, interval time.Duration, timeout time.Duration, host string) *HealthCheck { ctx, cancel := context.WithCancel(context.Background()) hc := &HealthCheck{ @@ -49,10 +57,13 @@ func NewHealthCheck(consumer HealthCheckConsumer, endpoint *url.URL, interval ti cancel: cancel, } - go hc.run() return hc } +func (hc *HealthCheck) Start() { + hc.start.Do(func() { go hc.run() }) +} + func (hc *HealthCheck) Close() { hc.cancel() } diff --git a/internal/server/load_balancer.go b/internal/server/load_balancer.go index 1c494ef6..a253ae7c 100644 --- a/internal/server/load_balancer.go +++ b/internal/server/load_balancer.go @@ -154,13 +154,14 @@ func (lb *LoadBalancer) MarkAllHealthy() { } func (lb *LoadBalancer) PrepareForWake() { - for _, target := range lb.all { - target.markUnhealthyForWake() - } lb.lock.Lock() lb.waitForHealthyContext, lb.markHealthy = context.WithCancel(context.Background()) lb.writers, lb.readers = TargetList{}, TargetList{} lb.lock.Unlock() + for _, target := range lb.all { + target.markUnhealthyForWake() + target.RestartHealthChecks() + } } func (lb *LoadBalancer) Dispose() { diff --git a/internal/server/load_balancer_test.go b/internal/server/load_balancer_test.go index 33e892c7..bc0c1a11 100644 --- a/internal/server/load_balancer_test.go +++ b/internal/server/load_balancer_test.go @@ -4,6 +4,7 @@ import ( "net/http" "net/http/httptest" "strconv" + "sync/atomic" "testing" "time" @@ -51,6 +52,34 @@ func TestLoadBalancer_WaitUntilHealthy(t *testing.T) { require.NoError(t, lb.WaitUntilHealthy(time.Second)) } +func TestLoadBalancer_PrepareForWakeRestartsSingleTargetHealthCheck(t *testing.T) { + var healthy atomic.Bool + healthy.Store(true) + _, targetURL := testBackendWithHandler(t, func(w http.ResponseWriter, r *http.Request) { + if !healthy.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + } + }) + options := defaultTargetOptions + options.HealthCheckConfig.Interval = time.Millisecond + target, err := NewTarget(targetURL, options) + require.NoError(t, err) + lb := NewLoadBalancer(TargetList{target}, DefaultWriterAffinityTimeout, false) + t.Cleanup(lb.Dispose) + require.NoError(t, lb.WaitUntilHealthy(time.Second)) + require.Eventually(t, func() bool { + target.inflightLock.Lock() + defer target.inflightLock.Unlock() + return target.healthcheck == nil + }, time.Second, time.Millisecond) + + healthy.Store(false) + lb.PrepareForWake() + time.AfterFunc(10*time.Millisecond, func() { healthy.Store(true) }) + require.NoError(t, lb.WaitUntilHealthy(time.Second)) + assert.Equal(t, TargetStateHealthy, target.State()) +} + func TestLoadBalancer_StartRequest(t *testing.T) { lb := testLoadBalancerWithHandlers(t, func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/server/target.go b/internal/server/target.go index 4faac43e..7bcf60bd 100644 --- a/internal/server/target.go +++ b/internal/server/target.go @@ -225,16 +225,23 @@ WAIT_FOR_REQUESTS_TO_COMPLETE: func (t *Target) BeginHealthChecks(stateConsumer TargetStateConsumer) { t.stateConsumer = stateConsumer + t.RestartHealthChecks() +} +func (t *Target) RestartHealthChecks() { t.withInflightLock(func() { + if t.healthcheck != nil { + t.healthcheck.Close() + } healthCheckURL := t.buildHealthCheckURL() - t.healthcheck = NewHealthCheck( + t.healthcheck = newHealthCheck( t, healthCheckURL, t.options.HealthCheckConfig.Interval, t.options.HealthCheckConfig.Timeout, t.options.HealthCheckConfig.Host, ) + t.healthcheck.Start() }) } From 093a3c98a581f17d22b86e22377ae242860801e1 Mon Sep 17 00:00:00 2001 From: komagata Date: Thu, 16 Jul 2026 04:11:55 +0900 Subject: [PATCH 06/17] Negotiate Docker API version --- README.md | 6 ++ internal/server/docker_client.go | 114 ++++++++++++++++++--- internal/server/docker_client_test.go | 136 +++++++++++++++++++++----- 3 files changed, 216 insertions(+), 40 deletions(-) diff --git a/README.md b/README.md index eed703b9..4d43b673 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,12 @@ enable this feature where that trust is acceptable; the lifecycle calls are isolated behind the `ContainerLifecycle` interface so they can be moved to an external service later. +The Docker client negotiates the API version once from the daemon's unversioned +`/version` endpoint and caches it for start/stop calls. If that endpoint is +unavailable or returns a non-success status, it falls back to the legacy +`v1.41` paths for compatibility with restricted socket proxies; a successful +but malformed version response is rejected instead of guessing. + The `deploy` command also waits for traffic to drain from the old instance before returning. This means it's safe to remove the old instance as soon as `deploy` returns successfully, without interrupting any in-flight requests. diff --git a/internal/server/docker_client.go b/internal/server/docker_client.go index fc087faf..96223ba8 100644 --- a/internal/server/docker_client.go +++ b/internal/server/docker_client.go @@ -2,13 +2,28 @@ package server import ( "context" + "encoding/json" "fmt" + "io" "net" "net/http" + "net/url" + "strconv" + "strings" + "sync" +) + +const ( + legacyDockerAPIVersion = "1.41" + maxDockerErrorBody = 4096 ) type DockerClient struct { httpClient *http.Client + + versionOnce sync.Once + apiVersion string + versionErr error } func NewDockerClient(socketPath string) *DockerClient { @@ -24,41 +39,110 @@ func NewDockerClient(socketPath string) *DockerClient { } func (c *DockerClient) StopContainer(ctx context.Context, name string) error { - url := fmt.Sprintf("http://localhost/v1.41/containers/%s/stop", name) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + return c.containerAction(ctx, name, "stop") +} + +func (c *DockerClient) StartContainer(ctx context.Context, name string) error { + return c.containerAction(ctx, name, "start") +} + +func (c *DockerClient) containerAction(ctx context.Context, name, action string) error { + version, err := c.negotiatedVersion(ctx) + if err != nil { + return err + } + endpoint := fmt.Sprintf("http://localhost/v%s/containers/%s/%s", version, url.PathEscape(name), action) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil) if err != nil { return err } - resp, err := c.httpClient.Do(req) if err != nil { return err } defer resp.Body.Close() - if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotModified { - return fmt.Errorf("unexpected status code from docker stop: %d", resp.StatusCode) + return dockerResponseError(action, resp) } - return nil } -func (c *DockerClient) StartContainer(ctx context.Context, name string) error { - url := fmt.Sprintf("http://localhost/v1.41/containers/%s/start", name) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) +func (c *DockerClient) negotiatedVersion(ctx context.Context) (string, error) { + c.versionOnce.Do(func() { + c.apiVersion, c.versionErr = c.queryVersion(ctx) + }) + return c.apiVersion, c.versionErr +} + +func (c *DockerClient) queryVersion(ctx context.Context) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost/version", nil) if err != nil { - return err + return "", err } - resp, err := c.httpClient.Do(req) if err != nil { - return err + // Some compatible Docker proxies do not expose /version. Preserve the + // legacy behavior and let the versioned operation return the useful error. + return legacyDockerAPIVersion, nil } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return legacyDockerAPIVersion, nil + } + var version struct { + APIVersion string `json:"ApiVersion"` + MinAPIVersion string `json:"MinAPIVersion"` + } + if err := json.NewDecoder(io.LimitReader(resp.Body, maxDockerErrorBody+1)).Decode(&version); err != nil { + return "", fmt.Errorf("invalid docker /version response: %w", err) + } + api, err := parseDockerAPIVersion(version.APIVersion) + if err != nil { + return "", fmt.Errorf("invalid docker ApiVersion: %w", err) + } + if version.MinAPIVersion != "" { + minimum, err := parseDockerAPIVersion(version.MinAPIVersion) + if err != nil { + return "", fmt.Errorf("invalid docker MinAPIVersion: %w", err) + } + if api < minimum { + return "", fmt.Errorf("invalid docker API range: ApiVersion %q is below MinAPIVersion %q", version.APIVersion, version.MinAPIVersion) + } + } + return version.APIVersion, nil +} - if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotModified { - return fmt.Errorf("unexpected status code from docker start: %d", resp.StatusCode) +func parseDockerAPIVersion(value string) (int, error) { + parts := strings.Split(value, ".") + if len(parts) != 2 { + return 0, fmt.Errorf("expected major.minor, got %q", value) } + major, err := strconv.Atoi(parts[0]) + if err != nil || major < 0 { + return 0, fmt.Errorf("invalid major version %q", parts[0]) + } + minor, err := strconv.Atoi(parts[1]) + if err != nil || minor < 0 { + return 0, fmt.Errorf("invalid minor version %q", parts[1]) + } + return major*1_000_000 + minor, nil +} - return nil +func dockerResponseError(action string, resp *http.Response) error { + body, err := io.ReadAll(io.LimitReader(resp.Body, maxDockerErrorBody+1)) + if err != nil { + return fmt.Errorf("docker %s returned status %d (reading error body: %w)", action, resp.StatusCode, err) + } + truncated := len(body) > maxDockerErrorBody + if truncated { + body = body[:maxDockerErrorBody] + } + message := strings.TrimSpace(string(body)) + if message == "" { + return fmt.Errorf("docker %s returned status %d", action, resp.StatusCode) + } + if truncated { + message += "…" + } + return fmt.Errorf("docker %s returned status %d: %s", action, resp.StatusCode, message) } diff --git a/internal/server/docker_client_test.go b/internal/server/docker_client_test.go index e3c5b4c6..09f08b34 100644 --- a/internal/server/docker_client_test.go +++ b/internal/server/docker_client_test.go @@ -2,43 +2,129 @@ package server import ( "context" + "fmt" "net" "net/http" "net/http/httptest" + "strings" + "sync" + "sync/atomic" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -func TestDockerClient_StopStart(t *testing.T) { - stopCalled := false - startCalled := false - - server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path == "/v1.41/containers/test-container/stop" { - stopCalled = true - w.WriteHeader(http.StatusNoContent) - } else if r.URL.Path == "/v1.41/containers/test-container/start" { - startCalled = true - w.WriteHeader(http.StatusNoContent) +func TestDockerClientNegotiatesNewDaemonAndUsesStartStopPaths(t *testing.T) { + var pathsMu sync.Mutex + var paths []string + client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { + pathsMu.Lock() + paths = append(paths, r.URL.Path) + pathsMu.Unlock() + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.52","MinAPIVersion":"1.44"}`) + return } - })) + w.WriteHeader(http.StatusNoContent) + }) - socketPath := t.TempDir() + "/docker.sock" - l, err := net.Listen("unix", socketPath) - require.NoError(t, err) - server.Listener = l - server.Start() - defer server.Close() + require.NoError(t, client.StopContainer(context.Background(), "test-container")) + require.NoError(t, client.StartContainer(context.Background(), "test-container")) + assert.Equal(t, []string{ + "/version", + "/v1.52/containers/test-container/stop", + "/v1.52/containers/test-container/start", + }, paths) +} + +func TestDockerClientNegotiatesOldCompatibleDaemon(t *testing.T) { + client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.41","MinAPIVersion":"1.24"}`) + return + } + assert.Equal(t, "/v1.41/containers/web/stop", r.URL.Path) + w.WriteHeader(http.StatusNoContent) + }) + require.NoError(t, client.StopContainer(context.Background(), "web")) +} + +func TestDockerClientFallsBackWhenVersionEndpointFails(t *testing.T) { + client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + http.Error(w, "not exposed", http.StatusNotFound) + return + } + assert.Equal(t, "/v1.41/containers/web/start", r.URL.Path) + w.WriteHeader(http.StatusNoContent) + }) + require.NoError(t, client.StartContainer(context.Background(), "web")) +} - client := NewDockerClient(socketPath) +func TestDockerClientRejectsMalformedVersionResponse(t *testing.T) { + for name, body := range map[string]string{ + "malformed json": `{`, + "invalid version": `{"ApiVersion":"new","MinAPIVersion":"1.44"}`, + "invalid range": `{"ApiVersion":"1.41","MinAPIVersion":"1.44"}`, + } { + t.Run(name, func(t *testing.T) { + client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, body) }) + err := client.StartContainer(context.Background(), "web") + require.Error(t, err) + assert.Contains(t, err.Error(), "docker") + }) + } +} + +func TestDockerClientNegotiatesOnlyOnceWithConcurrentFirstUse(t *testing.T) { + var versionCalls atomic.Int32 + client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + versionCalls.Add(1) + fmt.Fprint(w, `{"ApiVersion":"1.52","MinAPIVersion":"1.44"}`) + return + } + assert.True(t, strings.HasPrefix(r.URL.Path, "/v1.52/containers/")) + w.WriteHeader(http.StatusNoContent) + }) + + var wg sync.WaitGroup + for range 20 { + wg.Add(1) + go func() { + defer wg.Done() + require.NoError(t, client.StartContainer(context.Background(), "web")) + }() + } + wg.Wait() + assert.Equal(t, int32(1), versionCalls.Load()) +} - err = client.StopContainer(context.Background(), "test-container") - assert.NoError(t, err) - assert.True(t, stopCalled) +func TestDockerClientIncludesBoundedDockerErrorBody(t *testing.T) { + client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + fmt.Fprint(w, `{"ApiVersion":"1.52","MinAPIVersion":"1.44"}`) + return + } + w.WriteHeader(http.StatusBadRequest) + fmt.Fprint(w, "minimum API 1.44: "+strings.Repeat("x", maxDockerErrorBody*2)) + }) + err := client.StopContainer(context.Background(), "web") + require.Error(t, err) + assert.Contains(t, err.Error(), "minimum API 1.44") + assert.LessOrEqual(t, len(err.Error()), maxDockerErrorBody+100) + assert.True(t, strings.HasSuffix(err.Error(), "…")) +} - err = client.StartContainer(context.Background(), "test-container") - assert.NoError(t, err) - assert.True(t, startCalled) +func testDockerClient(t *testing.T, handler http.HandlerFunc) *DockerClient { + t.Helper() + server := httptest.NewUnstartedServer(handler) + socketPath := t.TempDir() + "/docker.sock" + listener, err := net.Listen("unix", socketPath) + require.NoError(t, err) + server.Listener = listener + server.Start() + t.Cleanup(server.Close) + return NewDockerClient(socketPath) } From 10cbca568f788ba39344c3a5675aa76c2898ca77 Mon Sep 17 00:00:00 2001 From: komagata Date: Sun, 19 Jul 2026 04:27:03 +0900 Subject: [PATCH 07/17] Simplify Docker API negotiation --- idle-controller-report.md | 82 --------------------------- internal/server/docker_client.go | 35 ++---------- internal/server/docker_client_test.go | 5 +- 3 files changed, 6 insertions(+), 116 deletions(-) delete mode 100644 idle-controller-report.md diff --git a/idle-controller-report.md b/idle-controller-report.md deleted file mode 100644 index 08c7f2ad..00000000 --- a/idle-controller-report.md +++ /dev/null @@ -1,82 +0,0 @@ -# Proxy idle controller implementation report - -## Assumptions and success criteria - -- Scale-to-zero is explicitly enabled per service only when `--idle-timeout` is greater than zero. -- Each write target hostname is also its Docker container name; ports are stripped before lifecycle calls. -- The proxy process can access the configured Docker Unix socket when the feature is enabled. -- Success means: never stop with an application request in flight; serialize stop versus arriving requests; coalesce concurrent wakes; hold requests until readiness, failure, cancellation, or timeout; preserve sleeping state over restart; and keep timeout-zero behavior unchanged. -- Health-check requests neither reset idle time nor wake a sleeping service. Paused/stopped services disable idle transitions; resume restarts the idle timer; deploy replaces the lifecycle target set and returns the controller to active. - -## CLI contract - -```text -kamal-proxy run [--docker-socket PATH] - --docker-socket default: /var/run/docker.sock - environment: DOCKER_SOCKET or KAMAL_PROXY_DOCKER_SOCKET - -kamal-proxy deploy SERVICE --target CONTAINER:PORT \ - [--idle-timeout DURATION] [--idle-wake-timeout DURATION] - --idle-timeout default: 0 (disabled) - --idle-wake-timeout default: 30s when idle is enabled - negative idle durations: rejected -``` - -The request is held before its body is read, so POST bodies are unchanged. WebSockets and streaming responses remain in flight until their handler returns and therefore prevent sleep; a newly arriving WebSocket/stream waits through wake like an ordinary request. - -## Implementation - -- `ContainerLifecycle` is the narrow start/stop interface. `DockerClient` is its Unix-socket implementation and can later be replaced by an external lifecycle client without changing the controller. -- The controller uses active/stopping/sleeping/waking transitions. A stopping barrier closes the race between the zero-inflight check and Docker stop; concurrent arrivals then share a single wake result. -- Start success is not enough: held requests wait for the active load balancer to observe a healthy target. Start, readiness, cancellation, and timeout errors release all waiters without consuming request bodies. -- Idle state is serialized in the existing state file. Incomplete stopping/waking transitions recover as sleeping and wake on the next application request. State transitions trigger serialized snapshots. - -## Changed files - -- `README.md` -- `internal/cmd/deploy.go` -- `internal/cmd/run.go` -- `internal/cmd/util.go` -- `internal/server/config.go` -- `internal/server/docker_client.go` -- `internal/server/docker_client_test.go` -- `internal/server/idle_controller.go` -- `internal/server/idle_controller_test.go` -- `internal/server/load_balancer.go` -- `internal/server/router.go` -- `internal/server/router_test.go` -- `internal/server/service.go` -- `internal/server/service_test.go` -- `internal/server/target.go` -- `internal/server/testing.go` - -Local commits: `105c829` (adapted PR #197) and `5d838ea` (race/failure/restart hardening and documentation). Nothing was pushed and no PR was opened. - -## Verification - -- `go test ./...` — pass -- `go test -race ./internal/server ./internal/cmd` — pass -- `go vet ./...` — pass -- `git diff --check` — pass -- `golangci-lint run` — not run; executable is not installed (`command not found`) - -Tests cover no-stop-with-inflight, stop completion versus arriving requests, concurrent wake coalescing, start failure, wake timeout, and persisted incomplete-state recovery. - -## Unresolved security and maintainer decisions - -- Mounting Docker's socket grants powerful host/container control. Maintainers must decide whether direct socket access is acceptable, whether a restricted socket proxy/external lifecycle service is required, and how container names should be authorized per service. -- The minimal client pins Docker API `v1.41`; maintainers should decide whether API negotiation is required for supported Docker versions. -- Only write targets are stopped. Reader targets and rollout targets remain running; expanding lifecycle scope needs an explicit product decision. -- Multi-container stop is sequential and not transactional. A later stop failure can leave earlier containers stopped; the controller returns active and logs the failure. Desired rollback/reconciliation semantics need a maintainer decision. -- Persisted sleeping state trusts that Docker state still matches the snapshot. The next request issues idempotent starts, but there is no startup reconciliation/list permission. - -## Recommended real-Docker integration scenarios - -1. Single container: idle stop, POST wake with a large/chunked body, health readiness, exact response verification. -2. Twenty concurrent GET/POST requests against one sleeping service: assert one Docker start and all bodies/results preserved. -3. Long SSE response and WebSocket: assert no stop while connected, then stop after close plus idle duration. -4. Restart kamal-proxy while sleeping and during a forced stopping/waking interruption; assert the next request recovers. -5. Pause/stop/resume and deploy while active, sleeping, and waking; verify no stale container is started and the new target set wins. -6. Docker socket unavailable, permission denied, start/stop HTTP errors, slow Docker response, and container missing/already started/stopped. -7. Container starts but health never succeeds, succeeds near the deadline, or flaps; verify waiter status and retry behavior. -8. Multiple write targets with a partial stop/start failure to decide and validate reconciliation semantics. diff --git a/internal/server/docker_client.go b/internal/server/docker_client.go index 96223ba8..3abda318 100644 --- a/internal/server/docker_client.go +++ b/internal/server/docker_client.go @@ -3,12 +3,12 @@ package server import ( "context" "encoding/json" + "errors" "fmt" "io" "net" "net/http" "net/url" - "strconv" "strings" "sync" ) @@ -90,44 +90,17 @@ func (c *DockerClient) queryVersion(ctx context.Context) (string, error) { return legacyDockerAPIVersion, nil } var version struct { - APIVersion string `json:"ApiVersion"` - MinAPIVersion string `json:"MinAPIVersion"` + APIVersion string `json:"ApiVersion"` } if err := json.NewDecoder(io.LimitReader(resp.Body, maxDockerErrorBody+1)).Decode(&version); err != nil { return "", fmt.Errorf("invalid docker /version response: %w", err) } - api, err := parseDockerAPIVersion(version.APIVersion) - if err != nil { - return "", fmt.Errorf("invalid docker ApiVersion: %w", err) - } - if version.MinAPIVersion != "" { - minimum, err := parseDockerAPIVersion(version.MinAPIVersion) - if err != nil { - return "", fmt.Errorf("invalid docker MinAPIVersion: %w", err) - } - if api < minimum { - return "", fmt.Errorf("invalid docker API range: ApiVersion %q is below MinAPIVersion %q", version.APIVersion, version.MinAPIVersion) - } + if version.APIVersion == "" { + return "", errors.New("docker /version response has no ApiVersion") } return version.APIVersion, nil } -func parseDockerAPIVersion(value string) (int, error) { - parts := strings.Split(value, ".") - if len(parts) != 2 { - return 0, fmt.Errorf("expected major.minor, got %q", value) - } - major, err := strconv.Atoi(parts[0]) - if err != nil || major < 0 { - return 0, fmt.Errorf("invalid major version %q", parts[0]) - } - minor, err := strconv.Atoi(parts[1]) - if err != nil || minor < 0 { - return 0, fmt.Errorf("invalid minor version %q", parts[1]) - } - return major*1_000_000 + minor, nil -} - func dockerResponseError(action string, resp *http.Response) error { body, err := io.ReadAll(io.LimitReader(resp.Body, maxDockerErrorBody+1)) if err != nil { diff --git a/internal/server/docker_client_test.go b/internal/server/docker_client_test.go index 09f08b34..2dd7f2a5 100644 --- a/internal/server/docker_client_test.go +++ b/internal/server/docker_client_test.go @@ -65,14 +65,13 @@ func TestDockerClientFallsBackWhenVersionEndpointFails(t *testing.T) { func TestDockerClientRejectsMalformedVersionResponse(t *testing.T) { for name, body := range map[string]string{ "malformed json": `{`, - "invalid version": `{"ApiVersion":"new","MinAPIVersion":"1.44"}`, - "invalid range": `{"ApiVersion":"1.41","MinAPIVersion":"1.44"}`, + "missing version": `{}`, } { t.Run(name, func(t *testing.T) { client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, body) }) err := client.StartContainer(context.Background(), "web") require.Error(t, err) - assert.Contains(t, err.Error(), "docker") + assert.Contains(t, err.Error(), "version") }) } } From c1587540a592a11bff2b4636a2693cf929fe8553 Mon Sep 17 00:00:00 2001 From: komagata Date: Tue, 21 Jul 2026 00:58:16 +0900 Subject: [PATCH 08/17] Update router tests for current main --- internal/server/router_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/server/router_test.go b/internal/server/router_test.go index 650e91ae..99f0431a 100644 --- a/internal/server/router_test.go +++ b/internal/server/router_test.go @@ -824,10 +824,10 @@ func TestRouter_RestoreLastSavedState_TLSOnDemandURL(t *testing.T) { serviceOptions.TLSEnabled = true serviceOptions.TLSOnDemandURL = allowServer.URL - router := NewRouter(statePath) + router := NewRouter(statePath, DefaultDockerSocketPath) require.NoError(t, router.DeployService("ondemand", []string{target}, defaultEmptyReaders, serviceOptions, defaultTargetOptions, defaultDeploymentOptions)) - router = NewRouter(statePath) + router = NewRouter(statePath, DefaultDockerSocketPath) require.NoError(t, router.RestoreLastSavedState()) service := router.services.Get("ondemand") From 32f0432b5e0a40f87e6f71d172ccc52cdd24d903 Mon Sep 17 00:00:00 2001 From: komagata Date: Tue, 21 Jul 2026 22:04:42 +0900 Subject: [PATCH 09/17] Address lifecycle review feedback --- internal/server/idle_controller.go | 83 ++++++++++++++++++------- internal/server/idle_controller_test.go | 46 ++++++++++++++ internal/server/load_balancer.go | 1 + internal/server/load_balancer_test.go | 14 +++++ internal/server/service.go | 3 + internal/server/service_test.go | 10 +++ 6 files changed, 135 insertions(+), 22 deletions(-) diff --git a/internal/server/idle_controller.go b/internal/server/idle_controller.go index fc98acfa..c494d8c2 100644 --- a/internal/server/idle_controller.go +++ b/internal/server/idle_controller.go @@ -46,18 +46,19 @@ type IdleController struct { WakeTimeout time.Duration `json:"wake_timeout"` ContainerNames []string `json:"container_names"` - mu sync.Mutex - lifecycle ContainerLifecycle - ready func(time.Duration) error - inflight int - lastRequest time.Time - wakeDone chan struct{} - wakeErr error - changed chan struct{} - closed chan struct{} - disabled bool - closeOnce sync.Once - persist func() + mu sync.Mutex + lifecycle ContainerLifecycle + ready func(time.Duration) error + inflight int + lastRequest time.Time + wakeDone chan struct{} + wakeErr error + lifecycleCancel context.CancelFunc + changed chan struct{} + closed chan struct{} + disabled bool + closeOnce sync.Once + persist func() } func NewIdleController(idleTimeout, wakeTimeout time.Duration, names []string, lifecycle ContainerLifecycle, ready func(time.Duration) error) *IdleController { @@ -155,7 +156,16 @@ func (c *IdleController) notifyPersist() { } } -func (c *IdleController) Disable() { c.mu.Lock(); c.disabled = true; c.mu.Unlock(); c.signal() } +func (c *IdleController) Disable() { + c.mu.Lock() + c.disabled = true + if c.State == IdleStateWaking || c.State == IdleStateStopping { + c.cancelLifecycleLocked() + c.State = IdleStateSleeping + } + c.mu.Unlock() + c.signal() +} func (c *IdleController) Enable() { c.mu.Lock() c.disabled = false @@ -169,6 +179,16 @@ func (c *IdleController) Reset(names []string, ready func(time.Duration) error) c.ContainerNames = append([]string(nil), names...) c.ready = ready c.State, c.wakeErr, c.lastRequest = IdleStateActive, nil, time.Now() + c.cancelLifecycleLocked() + c.mu.Unlock() + c.signal() +} + +func (c *IdleController) cancelLifecycleLocked() { + if c.lifecycleCancel != nil { + c.lifecycleCancel() + c.lifecycleCancel = nil + } if c.wakeDone != nil { select { case <-c.wakeDone: @@ -177,12 +197,13 @@ func (c *IdleController) Reset(names []string, ready func(time.Duration) error) } c.wakeDone = nil } - c.mu.Unlock() - c.signal() } func (c *IdleController) Close() { c.closeOnce.Do(func() { + c.mu.Lock() + c.cancelLifecycleLocked() + c.mu.Unlock() if c.closed != nil { close(c.closed) } @@ -233,23 +254,30 @@ func (c *IdleController) trySleep() { names, lifecycle := append([]string(nil), c.ContainerNames...), c.lifecycle c.State, c.wakeDone = IdleStateStopping, make(chan struct{}) stopDone := c.wakeDone - c.mu.Unlock() ctx, cancel := context.WithTimeout(context.Background(), DefaultIdleLifecycleTimeout) + c.lifecycleCancel = cancel + c.mu.Unlock() defer cancel() for _, name := range names { if err := lifecycle.StopContainer(ctx, name); err != nil { slog.Error("Failed to stop idle container", "container", name, "error", err) c.mu.Lock() - c.State, c.lastRequest = IdleStateActive, time.Now() - close(stopDone) + if c.wakeDone == stopDone { + c.State, c.lastRequest = IdleStateActive, time.Now() + c.lifecycleCancel = nil + close(stopDone) + } c.mu.Unlock() c.notifyPersist() return } } c.mu.Lock() - c.State = IdleStateSleeping - close(stopDone) + if c.wakeDone == stopDone { + c.State = IdleStateSleeping + c.lifecycleCancel = nil + close(stopDone) + } c.mu.Unlock() c.notifyPersist() c.signal() @@ -258,21 +286,32 @@ func (c *IdleController) trySleep() { func (c *IdleController) startWakeLocked() { c.State, c.wakeErr, c.wakeDone = IdleStateWaking, nil, make(chan struct{}) done, names, timeout, lifecycle, ready := c.wakeDone, append([]string(nil), c.ContainerNames...), c.WakeTimeout, c.lifecycle, c.ready + ctx, cancel := context.WithTimeout(context.Background(), timeout) + c.lifecycleCancel = cancel go func() { - ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() var err error for _, name := range names { + if ctx.Err() != nil { + err = ctx.Err() + break + } if err = lifecycle.StartContainer(ctx, name); err != nil { break } } if err == nil { - err = ready(timeout) + c.mu.Lock() + current := c.wakeDone == done + c.mu.Unlock() + if current { + err = ready(timeout) + } } c.mu.Lock() if c.wakeDone == done { c.wakeErr = err + c.lifecycleCancel = nil if err == nil { c.State, c.lastRequest = IdleStateActive, time.Now() } else { diff --git a/internal/server/idle_controller_test.go b/internal/server/idle_controller_test.go index ab213711..f3ea3dd7 100644 --- a/internal/server/idle_controller_test.go +++ b/internal/server/idle_controller_test.go @@ -18,6 +18,22 @@ type fakeLifecycle struct { startErr, stopErr error } +type blockingLifecycle struct { + starts atomic.Int32 + started chan struct{} + finished chan struct{} +} + +func (f *blockingLifecycle) StartContainer(ctx context.Context, _ string) error { + f.starts.Add(1) + close(f.started) + <-ctx.Done() + close(f.finished) + return ctx.Err() +} + +func (f *blockingLifecycle) StopContainer(context.Context, string) error { return nil } + func (f *fakeLifecycle) StartContainer(context.Context, string) error { f.starts.Add(1) return f.startErr @@ -93,3 +109,33 @@ func TestIdleControllerRestoresSleepingAndWakes(t *testing.T) { restored.EndRequest() assert.Equal(t, int32(1), lifecycle.starts.Load()) } + +func TestIdleControllerResetCancelsWake(t *testing.T) { + lifecycle := &blockingLifecycle{started: make(chan struct{}), finished: make(chan struct{})} + readyCalled := atomic.Bool{} + c := NewIdleController(time.Hour, time.Second, []string{"old"}, lifecycle, func(time.Duration) error { + readyCalled.Store(true) + return nil + }) + defer c.Close() + c.mu.Lock() + c.State = IdleStateSleeping + c.mu.Unlock() + + done := make(chan error, 1) + go func() { done <- c.BeginRequest(context.Background()) }() + <-lifecycle.started + c.Reset([]string{"new"}, func(time.Duration) error { return nil }) + + require.NoError(t, <-done) + require.Eventually(t, func() bool { + select { + case <-lifecycle.finished: + return true + default: + return false + } + }, 100*time.Millisecond, time.Millisecond) + assert.False(t, readyCalled.Load()) + assert.Equal(t, IdleStateActive, c.StateValue()) +} diff --git a/internal/server/load_balancer.go b/internal/server/load_balancer.go index a253ae7c..a4d20267 100644 --- a/internal/server/load_balancer.go +++ b/internal/server/load_balancer.go @@ -155,6 +155,7 @@ func (lb *LoadBalancer) MarkAllHealthy() { func (lb *LoadBalancer) PrepareForWake() { lb.lock.Lock() + lb.markHealthy() lb.waitForHealthyContext, lb.markHealthy = context.WithCancel(context.Background()) lb.writers, lb.readers = TargetList{}, TargetList{} lb.lock.Unlock() diff --git a/internal/server/load_balancer_test.go b/internal/server/load_balancer_test.go index bc0c1a11..e7615be0 100644 --- a/internal/server/load_balancer_test.go +++ b/internal/server/load_balancer_test.go @@ -80,6 +80,20 @@ func TestLoadBalancer_PrepareForWakeRestartsSingleTargetHealthCheck(t *testing.T assert.Equal(t, TargetStateHealthy, target.State()) } +func TestLoadBalancer_PrepareForWakeReleasesPreviousWaiters(t *testing.T) { + lb := NewLoadBalancer(TargetList{}, DefaultWriterAffinityTimeout, false) + t.Cleanup(lb.Dispose) + previous := lb.waitForHealthyContext + + lb.PrepareForWake() + + select { + case <-previous.Done(): + case <-time.After(time.Second): + t.Fatal("previous health wait was not released") + } +} + func TestLoadBalancer_StartRequest(t *testing.T) { lb := testLoadBalancerWithHandlers(t, func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/server/service.go b/internal/server/service.go index 21aca9eb..02a12645 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -496,6 +496,9 @@ func (s *Service) waitUntilActiveHealthy(timeout time.Duration) error { s.serviceLock.RLock() lb := s.active s.serviceLock.RUnlock() + if lb == nil { + return ErrorNoHealthyTargets + } lb.PrepareForWake() return lb.WaitUntilHealthy(timeout) } diff --git a/internal/server/service_test.go b/internal/server/service_test.go index e7694c0b..129dad38 100644 --- a/internal/server/service_test.go +++ b/internal/server/service_test.go @@ -26,6 +26,16 @@ func TestService_ServeRequest(t *testing.T) { require.Equal(t, http.StatusOK, w.Result().StatusCode) } +func TestService_WaitUntilActiveHealthyWithoutLoadBalancer(t *testing.T) { + options := defaultServiceOptions + options.IdleTimeout = time.Minute + service, err := NewService("test", options, defaultTargetOptions, &fakeLifecycle{}) + require.NoError(t, err) + t.Cleanup(service.idleController.Close) + + assert.ErrorIs(t, service.waitUntilActiveHealthy(time.Second), ErrorNoHealthyTargets) +} + func TestService_ClientIPHeaderRewritesXForwardedFor(t *testing.T) { var xForwardedFor, trueClientIP string From bf08b5a9cf20b99751812546ef54ae0fdabd18b4 Mon Sep 17 00:00:00 2001 From: komagata Date: Tue, 21 Jul 2026 22:25:44 +0900 Subject: [PATCH 10/17] Address state handling review feedback --- internal/server/idle_controller.go | 17 +++++++++++++++++ internal/server/idle_controller_test.go | 18 ++++++++++++++++++ internal/server/router.go | 10 ++++++++-- internal/server/router_test.go | 14 ++++++++++++++ internal/server/service.go | 2 +- internal/server/service_test.go | 21 +++++++++++++++++++++ 6 files changed, 79 insertions(+), 3 deletions(-) diff --git a/internal/server/idle_controller.go b/internal/server/idle_controller.go index c494d8c2..ee127ece 100644 --- a/internal/server/idle_controller.go +++ b/internal/server/idle_controller.go @@ -67,6 +67,23 @@ func NewIdleController(idleTimeout, wakeTimeout time.Duration, names []string, l return c } +func (c *IdleController) MarshalJSON() ([]byte, error) { + type persisted struct { + State IdleState `json:"state"` + IdleTimeout time.Duration `json:"idle_timeout"` + WakeTimeout time.Duration `json:"wake_timeout"` + ContainerNames []string `json:"container_names"` + } + c.mu.Lock() + defer c.mu.Unlock() + return json.Marshal(persisted{ + State: c.State, + IdleTimeout: c.IdleTimeout, + WakeTimeout: c.WakeTimeout, + ContainerNames: c.ContainerNames, + }) +} + func (c *IdleController) UnmarshalJSON(data []byte) error { type persisted struct { State IdleState `json:"state"` diff --git a/internal/server/idle_controller_test.go b/internal/server/idle_controller_test.go index f3ea3dd7..47864882 100644 --- a/internal/server/idle_controller_test.go +++ b/internal/server/idle_controller_test.go @@ -110,6 +110,24 @@ func TestIdleControllerRestoresSleepingAndWakes(t *testing.T) { assert.Equal(t, int32(1), lifecycle.starts.Load()) } +func TestIdleControllerMarshalWaitsForStateLock(t *testing.T) { + c := &IdleController{State: IdleStateSleeping} + c.mu.Lock() + done := make(chan error, 1) + go func() { + _, err := json.Marshal(c) + done <- err + }() + + select { + case <-done: + t.Fatal("MarshalJSON read state without acquiring the lock") + case <-time.After(10 * time.Millisecond): + } + c.mu.Unlock() + require.NoError(t, <-done) +} + func TestIdleControllerResetCancelsWake(t *testing.T) { lifecycle := &blockingLifecycle{started: make(chan struct{}), finished: make(chan struct{})} readyCalled := atomic.Bool{} diff --git a/internal/server/router.go b/internal/server/router.go index 15bf805c..5cdf6c3c 100644 --- a/internal/server/router.go +++ b/internal/server/router.go @@ -95,7 +95,7 @@ func (r *Router) RestoreLastSavedState() error { r.services = NewServiceMap() for _, service := range services { service.lifecycle = r.dockerClient - service.stateChanged = func() { _ = r.saveStateSnapshot() } + service.stateChanged = r.saveStateSnapshotWithLogging if err := service.initialize(service.options, service.targetOptions); err != nil { return err } @@ -320,7 +320,7 @@ func (r *Router) createOrUpdateService(name string, options ServiceOptions, targ if service == nil { service, err := NewService(name, options, targetOptions, r.dockerClient) if err == nil { - service.stateChanged = func() { _ = r.saveStateSnapshot() } + service.stateChanged = r.saveStateSnapshotWithLogging if service.idleController != nil { service.idleController.SetPersist(service.stateChanged) } @@ -403,6 +403,12 @@ func (r *Router) saveStateSnapshot() error { return nil } +func (r *Router) saveStateSnapshotWithLogging() { + if err := r.saveStateSnapshot(); err != nil { + slog.Error("Unable to save state snapshot", "error", err, "path", r.statePath) + } +} + func (r *Router) serviceForRequest(req *http.Request) (*Service, string) { r.serviceLock.RLock() defer r.serviceLock.RUnlock() diff --git a/internal/server/router_test.go b/internal/server/router_test.go index 99f0431a..46c5f245 100644 --- a/internal/server/router_test.go +++ b/internal/server/router_test.go @@ -1,9 +1,11 @@ package server import ( + "bytes" "context" "crypto/tls" "encoding/json" + "log/slog" "net/http" "net/http/httptest" "os" @@ -26,6 +28,18 @@ func TestRouter_Empty(t *testing.T) { assert.Equal(t, http.StatusNotFound, statusCode) } +func TestRouter_StateChangedLogsSaveErrors(t *testing.T) { + var logs bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + t.Cleanup(func() { slog.SetDefault(previous) }) + router := NewRouter(t.TempDir(), DefaultDockerSocketPath) + + router.saveStateSnapshotWithLogging() + + assert.Contains(t, logs.String(), "Unable to save state snapshot") +} + func TestRouter_DeployService(t *testing.T) { router := testRouter(t) _, target := testBackend(t, "first", http.StatusOK) diff --git a/internal/server/service.go b/internal/server/service.go index 02a12645..0da4476f 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -302,7 +302,7 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { defer metrics.Tracker.SubtractInflightRequest(s.name) } - if s.idleController != nil && !s.targetOptions.IsHealthCheckRequest(r) { + if s.idleController != nil && s.pauseController.GetState() == PauseStateRunning && !s.targetOptions.IsHealthCheckRequest(r) { if err := s.idleController.BeginRequest(r.Context()); err != nil { s.idleController.EndRequest() SetErrorResponse(w, r, http.StatusServiceUnavailable, err) diff --git a/internal/server/service_test.go b/internal/server/service_test.go index 129dad38..68060b2b 100644 --- a/internal/server/service_test.go +++ b/internal/server/service_test.go @@ -26,6 +26,27 @@ func TestService_ServeRequest(t *testing.T) { require.Equal(t, http.StatusOK, w.Result().StatusCode) } +func TestService_StoppedRequestDoesNotWakeIdleContainer(t *testing.T) { + options := defaultServiceOptions + options.IdleTimeout = time.Minute + options.IdleWakeTimeout = time.Second + service := testCreateService(t, options, defaultTargetOptions) + t.Cleanup(service.Dispose) + lifecycle := &fakeLifecycle{} + service.idleController.mu.Lock() + service.idleController.lifecycle = lifecycle + service.idleController.State = IdleStateSleeping + service.idleController.mu.Unlock() + require.NoError(t, service.pauseController.Stop(DefaultStopMessage)) + + req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil) + w := httptest.NewRecorder() + service.ServeHTTP(w, req) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Zero(t, lifecycle.starts.Load()) +} + func TestService_WaitUntilActiveHealthyWithoutLoadBalancer(t *testing.T) { options := defaultServiceOptions options.IdleTimeout = time.Minute From 26d844155b4611e5b3a380472b5871caf000f34d Mon Sep 17 00:00:00 2001 From: komagata Date: Tue, 21 Jul 2026 22:40:37 +0900 Subject: [PATCH 11/17] Address persistence review feedback --- README.md | 2 +- internal/server/router.go | 13 +++++++++---- internal/server/service.go | 2 +- 3 files changed, 11 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 4d43b673..73e34147 100644 --- a/README.md +++ b/README.md @@ -69,7 +69,7 @@ unchanged after a successful wake. Concurrent wake requests are coalesced. Open streaming responses and WebSockets count as activity/in-flight work and prevent sleeping until they close; a new stream or WebSocket is held during wake like any other request. Health-check requests do not wake or reset an idle -service and receive success while it is sleeping. +service and receive success while it is stopping, sleeping, or waking. Mounting the Docker socket gives the proxy host-level container control. Only enable this feature where that trust is acceptable; the lifecycle calls are diff --git a/internal/server/router.go b/internal/server/router.go index 5cdf6c3c..7697b919 100644 --- a/internal/server/router.go +++ b/internal/server/router.go @@ -393,10 +393,15 @@ func (r *Router) saveStateSnapshot() error { return err } - err = json.NewEncoder(f).Encode(services) - if err != nil { - slog.Error("Unable to save state", "error", err, "path", r.statePath) - return err + encodeErr := json.NewEncoder(f).Encode(services) + closeErr := f.Close() + if encodeErr != nil { + slog.Error("Unable to save state", "error", encodeErr, "path", r.statePath) + return encodeErr + } + if closeErr != nil { + slog.Error("Unable to close saved state", "error", closeErr, "path", r.statePath) + return closeErr } slog.Debug("Saved state", "path", r.statePath) diff --git a/internal/server/service.go b/internal/server/service.go index 0da4476f..8d4e821e 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -633,7 +633,7 @@ func (s *Service) handleIdleHealthCheck(w http.ResponseWriter, r *http.Request) if s.targetOptions.IsHealthCheckRequest(r) { // Health checks should not wake the service. - // If it's sleeping, just return 200. + // While it is stopping, sleeping, or waking, just return 200. icState := s.idleController.StateValue() if icState != IdleStateActive { w.WriteHeader(http.StatusOK) From acbce5a273fdaf6e03164cb18f6db342e78be218 Mon Sep 17 00:00:00 2001 From: komagata Date: Wed, 22 Jul 2026 14:30:28 +0900 Subject: [PATCH 12/17] Address restore and wake timeout feedback --- internal/cmd/run.go | 4 +++- internal/cmd/run_test.go | 20 +++++++++++++++++ internal/server/idle_controller.go | 8 ++++++- internal/server/idle_controller_test.go | 30 ++++++++++++++++++++++++- internal/server/router.go | 3 ++- 5 files changed, 61 insertions(+), 4 deletions(-) create mode 100644 internal/cmd/run_test.go diff --git a/internal/cmd/run.go b/internal/cmd/run.go index 1c2689ea..0831455b 100644 --- a/internal/cmd/run.go +++ b/internal/cmd/run.go @@ -38,7 +38,9 @@ func (c *runCommand) run(cmd *cobra.Command, args []string) error { c.setLogger() router := server.NewRouter(globalConfig.StatePath(), globalConfig.DockerSocketPath) - router.RestoreLastSavedState() + if err := router.RestoreLastSavedState(); err != nil { + return err + } s := server.NewServer(&globalConfig, router) err := s.Start() diff --git a/internal/cmd/run_test.go b/internal/cmd/run_test.go new file mode 100644 index 00000000..3269eea5 --- /dev/null +++ b/internal/cmd/run_test.go @@ -0,0 +1,20 @@ +package cmd + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRunCommandReturnsStateRestoreError(t *testing.T) { + previous := globalConfig + t.Cleanup(func() { globalConfig = previous }) + globalConfig.AlternateConfigDir = t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(globalConfig.AlternateConfigDir, "kamal-proxy.state"), []byte("invalid"), 0o600)) + + err := newRunCommand().run(nil, nil) + + require.ErrorContains(t, err, "invalid character 'i'") +} diff --git a/internal/server/idle_controller.go b/internal/server/idle_controller.go index ee127ece..0c0f85c7 100644 --- a/internal/server/idle_controller.go +++ b/internal/server/idle_controller.go @@ -304,6 +304,7 @@ func (c *IdleController) startWakeLocked() { c.State, c.wakeErr, c.wakeDone = IdleStateWaking, nil, make(chan struct{}) done, names, timeout, lifecycle, ready := c.wakeDone, append([]string(nil), c.ContainerNames...), c.WakeTimeout, c.lifecycle, c.ready ctx, cancel := context.WithTimeout(context.Background(), timeout) + deadline, _ := ctx.Deadline() c.lifecycleCancel = cancel go func() { defer cancel() @@ -322,7 +323,12 @@ func (c *IdleController) startWakeLocked() { current := c.wakeDone == done c.mu.Unlock() if current { - err = ready(timeout) + remaining := time.Until(deadline) + if remaining <= 0 { + err = ErrIdleWakeTimeout + } else { + err = ready(remaining) + } } } c.mu.Lock() diff --git a/internal/server/idle_controller_test.go b/internal/server/idle_controller_test.go index 47864882..1aae42e0 100644 --- a/internal/server/idle_controller_test.go +++ b/internal/server/idle_controller_test.go @@ -16,6 +16,7 @@ import ( type fakeLifecycle struct { starts, stops atomic.Int32 startErr, stopErr error + startDelay time.Duration } type blockingLifecycle struct { @@ -34,8 +35,15 @@ func (f *blockingLifecycle) StartContainer(ctx context.Context, _ string) error func (f *blockingLifecycle) StopContainer(context.Context, string) error { return nil } -func (f *fakeLifecycle) StartContainer(context.Context, string) error { +func (f *fakeLifecycle) StartContainer(ctx context.Context, _ string) error { f.starts.Add(1) + if f.startDelay > 0 { + select { + case <-time.After(f.startDelay): + case <-ctx.Done(): + return ctx.Err() + } + } return f.startErr } func (f *fakeLifecycle) StopContainer(context.Context, string) error { @@ -157,3 +165,23 @@ func TestIdleControllerResetCancelsWake(t *testing.T) { assert.False(t, readyCalled.Load()) assert.Equal(t, IdleStateActive, c.StateValue()) } + +func TestIdleControllerWakeReadinessUsesRemainingTimeout(t *testing.T) { + wakeTimeout := 100 * time.Millisecond + lifecycle := &fakeLifecycle{startDelay: 10 * time.Millisecond} + readyTimeout := make(chan time.Duration, 1) + c := NewIdleController(time.Hour, wakeTimeout, []string{"web"}, lifecycle, func(timeout time.Duration) error { + readyTimeout <- timeout + return nil + }) + defer c.Close() + c.mu.Lock() + c.State = IdleStateSleeping + c.mu.Unlock() + + require.NoError(t, c.BeginRequest(context.Background())) + c.EndRequest() + remaining := <-readyTimeout + assert.Positive(t, remaining) + assert.Less(t, remaining, wakeTimeout) +} diff --git a/internal/server/router.go b/internal/server/router.go index 7697b919..783b06f8 100644 --- a/internal/server/router.go +++ b/internal/server/router.go @@ -5,6 +5,7 @@ import ( "crypto/tls" "encoding/json" "errors" + "fmt" "log/slog" "net/http" "os" @@ -97,7 +98,7 @@ func (r *Router) RestoreLastSavedState() error { service.lifecycle = r.dockerClient service.stateChanged = r.saveStateSnapshotWithLogging if err := service.initialize(service.options, service.targetOptions); err != nil { - return err + return fmt.Errorf("initialize restored service %q: %w", service.name, err) } r.services.Set(service) } From 33940a9bb0ffda2f30db5c0a17d85a4b7e425b6d Mon Sep 17 00:00:00 2001 From: komagata Date: Thu, 23 Jul 2026 00:48:22 +0900 Subject: [PATCH 13/17] Render idle wake errors in 503 responses --- internal/server/service.go | 3 ++- internal/server/service_test.go | 25 +++++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/internal/server/service.go b/internal/server/service.go index 8d4e821e..20e5e8b6 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -305,7 +305,8 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { if s.idleController != nil && s.pauseController.GetState() == PauseStateRunning && !s.targetOptions.IsHealthCheckRequest(r) { if err := s.idleController.BeginRequest(r.Context()); err != nil { s.idleController.EndRequest() - SetErrorResponse(w, r, http.StatusServiceUnavailable, err) + templateArguments := struct{ Message string }{err.Error()} + SetErrorResponse(w, r, http.StatusServiceUnavailable, templateArguments) return } defer s.idleController.EndRequest() diff --git a/internal/server/service_test.go b/internal/server/service_test.go index 68060b2b..9d0b00de 100644 --- a/internal/server/service_test.go +++ b/internal/server/service_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "net" "net/http" "net/http/httptest" @@ -14,6 +15,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/basecamp/kamal-proxy/internal/pages" ) func TestService_ServeRequest(t *testing.T) { @@ -47,6 +50,28 @@ func TestService_StoppedRequestDoesNotWakeIdleContainer(t *testing.T) { assert.Zero(t, lifecycle.starts.Load()) } +func TestService_WakeFailureRendersErrorMessage(t *testing.T) { + options := defaultServiceOptions + options.IdleTimeout = time.Minute + options.IdleWakeTimeout = time.Second + service := testCreateService(t, options, defaultTargetOptions) + t.Cleanup(service.Dispose) + lifecycle := &fakeLifecycle{startErr: errors.New("start failed")} + service.idleController.mu.Lock() + service.idleController.lifecycle = lifecycle + service.idleController.State = IdleStateSleeping + service.idleController.mu.Unlock() + + req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil) + w := httptest.NewRecorder() + handler, err := WithErrorPageMiddleware(pages.DefaultErrorPages, true, service) + require.NoError(t, err) + handler.ServeHTTP(w, req) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Contains(t, w.Body.String(), "start failed") +} + func TestService_WaitUntilActiveHealthyWithoutLoadBalancer(t *testing.T) { options := defaultServiceOptions options.IdleTimeout = time.Minute From 47daa3b0fd1f77342e40606966cd6477a4fce6e2 Mon Sep 17 00:00:00 2001 From: komagata Date: Thu, 23 Jul 2026 10:51:23 +0900 Subject: [PATCH 14/17] Centralize state save logging --- internal/server/router.go | 11 +++-------- internal/server/router_test.go | 2 +- internal/server/service_test.go | 2 +- 3 files changed, 5 insertions(+), 10 deletions(-) diff --git a/internal/server/router.go b/internal/server/router.go index 783b06f8..5158fb35 100644 --- a/internal/server/router.go +++ b/internal/server/router.go @@ -96,7 +96,7 @@ func (r *Router) RestoreLastSavedState() error { r.services = NewServiceMap() for _, service := range services { service.lifecycle = r.dockerClient - service.stateChanged = r.saveStateSnapshotWithLogging + service.stateChanged = func() { _ = r.saveStateSnapshot() } if err := service.initialize(service.options, service.targetOptions); err != nil { return fmt.Errorf("initialize restored service %q: %w", service.name, err) } @@ -321,7 +321,7 @@ func (r *Router) createOrUpdateService(name string, options ServiceOptions, targ if service == nil { service, err := NewService(name, options, targetOptions, r.dockerClient) if err == nil { - service.stateChanged = r.saveStateSnapshotWithLogging + service.stateChanged = func() { _ = r.saveStateSnapshot() } if service.idleController != nil { service.idleController.SetPersist(service.stateChanged) } @@ -391,6 +391,7 @@ func (r *Router) saveStateSnapshot() error { f, err := os.Create(r.statePath) if err != nil { + slog.Error("Unable to save state snapshot", "error", err, "path", r.statePath) return err } @@ -409,12 +410,6 @@ func (r *Router) saveStateSnapshot() error { return nil } -func (r *Router) saveStateSnapshotWithLogging() { - if err := r.saveStateSnapshot(); err != nil { - slog.Error("Unable to save state snapshot", "error", err, "path", r.statePath) - } -} - func (r *Router) serviceForRequest(req *http.Request) (*Service, string) { r.serviceLock.RLock() defer r.serviceLock.RUnlock() diff --git a/internal/server/router_test.go b/internal/server/router_test.go index 46c5f245..cae77339 100644 --- a/internal/server/router_test.go +++ b/internal/server/router_test.go @@ -35,7 +35,7 @@ func TestRouter_StateChangedLogsSaveErrors(t *testing.T) { t.Cleanup(func() { slog.SetDefault(previous) }) router := NewRouter(t.TempDir(), DefaultDockerSocketPath) - router.saveStateSnapshotWithLogging() + require.Error(t, router.saveStateSnapshot()) assert.Contains(t, logs.String(), "Unable to save state snapshot") } diff --git a/internal/server/service_test.go b/internal/server/service_test.go index 9d0b00de..5f3494f0 100644 --- a/internal/server/service_test.go +++ b/internal/server/service_test.go @@ -431,7 +431,7 @@ func testCreateServiceWithHandler(t *testing.T, options ServiceOptions, targetOp target, err := NewTarget(serverURL.Host, targetOptions) require.NoError(t, err) - service, err := NewService("test", options, targetOptions, NewDockerClient(DefaultDockerSocketPath)) + service, err := NewService("test", options, targetOptions, &fakeLifecycle{}) require.NoError(t, err) service.UpdateLoadBalancer(NewLoadBalancer(TargetList{target}, DefaultWriterAffinityTimeout, false), TargetSlotActive) From bc12779a5945c3d0e83f4ed30c1434fbd1e68159 Mon Sep 17 00:00:00 2001 From: komagata Date: Thu, 23 Jul 2026 12:13:05 +0900 Subject: [PATCH 15/17] Handle idle lifecycle during rollouts --- internal/server/service.go | 44 +++++++++++++++++++++++++++------ internal/server/service_test.go | 35 ++++++++++++++++++++++++-- 2 files changed, 69 insertions(+), 10 deletions(-) diff --git a/internal/server/service.go b/internal/server/service.go index 20e5e8b6..b5275bb5 100644 --- a/internal/server/service.go +++ b/internal/server/service.go @@ -261,11 +261,14 @@ func (s *Service) UpdateLoadBalancer(lb *LoadBalancer, slot TargetSlot) *LoadBal if slot == TargetSlotRollout { replaced = s.rollout s.rollout = lb + if s.idleController != nil { + s.idleController.configure(s.options.IdleTimeout, s.options.IdleWakeTimeout, s.idleContainerNames(), s.lifecycle, s.waitUntilIdleTargetsHealthy) + } } else { replaced = s.active s.active = lb if s.idleController != nil { - s.idleController.Reset(s.activeContainerNames(), s.waitUntilActiveHealthy) + s.idleController.Reset(s.idleContainerNames(), s.waitUntilIdleTargetsHealthy) } } @@ -469,9 +472,9 @@ func (s *Service) initialize(options ServiceOptions, targetOptions TargetOptions if s.options.IdleTimeout > 0 { if s.idleController == nil { - s.idleController = NewIdleController(s.options.IdleTimeout, s.options.IdleWakeTimeout, s.activeContainerNames(), s.lifecycle, s.waitUntilActiveHealthy) + s.idleController = NewIdleController(s.options.IdleTimeout, s.options.IdleWakeTimeout, s.idleContainerNames(), s.lifecycle, s.waitUntilIdleTargetsHealthy) } else { - s.idleController.configure(s.options.IdleTimeout, s.options.IdleWakeTimeout, s.activeContainerNames(), s.lifecycle, s.waitUntilActiveHealthy) + s.idleController.configure(s.options.IdleTimeout, s.options.IdleWakeTimeout, s.idleContainerNames(), s.lifecycle, s.waitUntilIdleTargetsHealthy) } s.idleController.SetPersist(s.stateChanged) } else if s.idleController != nil { @@ -493,15 +496,40 @@ func (s *Service) activeContainerNames() []string { return names } -func (s *Service) waitUntilActiveHealthy(timeout time.Duration) error { +func (s *Service) idleContainerNames() []string { + names := s.activeContainerNames() + if s.rollout != nil { + for _, target := range s.rollout.WriteTargets() { + names = append(names, target.ContainerName()) + } + } + return names +} + +func (s *Service) waitUntilIdleTargetsHealthy(timeout time.Duration) error { s.serviceLock.RLock() - lb := s.active + loadBalancers := make([]*LoadBalancer, 0, 2) + if s.active != nil { + loadBalancers = append(loadBalancers, s.active) + } + if s.rollout != nil { + loadBalancers = append(loadBalancers, s.rollout) + } s.serviceLock.RUnlock() - if lb == nil { + if len(loadBalancers) == 0 { return ErrorNoHealthyTargets } - lb.PrepareForWake() - return lb.WaitUntilHealthy(timeout) + errs := make(chan error, len(loadBalancers)) + for _, lb := range loadBalancers { + lb.PrepareForWake() + go func() { errs <- lb.WaitUntilHealthy(timeout) }() + } + for range loadBalancers { + if err := <-errs; err != nil { + return err + } + } + return nil } func (s *Service) Drain(timeout time.Duration) { diff --git a/internal/server/service_test.go b/internal/server/service_test.go index 5f3494f0..bb1d5016 100644 --- a/internal/server/service_test.go +++ b/internal/server/service_test.go @@ -72,14 +72,45 @@ func TestService_WakeFailureRendersErrorMessage(t *testing.T) { assert.Contains(t, w.Body.String(), "start failed") } -func TestService_WaitUntilActiveHealthyWithoutLoadBalancer(t *testing.T) { +func TestService_WakeStartsActiveAndRolloutTargets(t *testing.T) { + options := defaultServiceOptions + options.IdleTimeout = time.Minute + options.IdleWakeTimeout = time.Second + service := testCreateServiceWithHandler(t, options, defaultTargetOptions, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("active")) + })) + t.Cleanup(service.Dispose) + _, rolloutTarget := testBackend(t, "rollout", http.StatusOK) + target, err := NewTarget(rolloutTarget, defaultTargetOptions) + require.NoError(t, err) + rollout := NewLoadBalancer(TargetList{target}, DefaultWriterAffinityTimeout, false) + require.NoError(t, rollout.WaitUntilHealthy(time.Second)) + service.UpdateLoadBalancer(rollout, TargetSlotRollout) + require.NoError(t, service.SetRolloutSplit(100, nil)) + lifecycle := &fakeLifecycle{} + service.idleController.mu.Lock() + service.idleController.lifecycle = lifecycle + service.idleController.State = IdleStateSleeping + service.idleController.mu.Unlock() + + req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil) + req.AddCookie(&http.Cookie{Name: RolloutCookieName, Value: "1"}) + w := httptest.NewRecorder() + service.ServeHTTP(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "rollout", w.Body.String()) + assert.Equal(t, int32(2), lifecycle.starts.Load()) +} + +func TestService_WaitUntilIdleTargetsHealthyWithoutLoadBalancer(t *testing.T) { options := defaultServiceOptions options.IdleTimeout = time.Minute service, err := NewService("test", options, defaultTargetOptions, &fakeLifecycle{}) require.NoError(t, err) t.Cleanup(service.idleController.Close) - assert.ErrorIs(t, service.waitUntilActiveHealthy(time.Second), ErrorNoHealthyTargets) + assert.ErrorIs(t, service.waitUntilIdleTargetsHealthy(time.Second), ErrorNoHealthyTargets) } func TestService_ClientIPHeaderRewritesXForwardedFor(t *testing.T) { From 7d5eb882c5dcd3bd0b15f779038b8b1845efe976 Mon Sep 17 00:00:00 2001 From: komagata Date: Thu, 23 Jul 2026 12:38:11 +0900 Subject: [PATCH 16/17] Recover from partial idle stop failures --- internal/server/idle_controller.go | 2 +- internal/server/idle_controller_test.go | 24 +++++++++++++++++++++++- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/internal/server/idle_controller.go b/internal/server/idle_controller.go index 0c0f85c7..39f2e994 100644 --- a/internal/server/idle_controller.go +++ b/internal/server/idle_controller.go @@ -280,7 +280,7 @@ func (c *IdleController) trySleep() { slog.Error("Failed to stop idle container", "container", name, "error", err) c.mu.Lock() if c.wakeDone == stopDone { - c.State, c.lastRequest = IdleStateActive, time.Now() + c.State = IdleStateSleeping c.lifecycleCancel = nil close(stopDone) } diff --git a/internal/server/idle_controller_test.go b/internal/server/idle_controller_test.go index 1aae42e0..c6556228 100644 --- a/internal/server/idle_controller_test.go +++ b/internal/server/idle_controller_test.go @@ -17,6 +17,7 @@ type fakeLifecycle struct { starts, stops atomic.Int32 startErr, stopErr error startDelay time.Duration + stopErrAt int32 } type blockingLifecycle struct { @@ -47,7 +48,10 @@ func (f *fakeLifecycle) StartContainer(ctx context.Context, _ string) error { return f.startErr } func (f *fakeLifecycle) StopContainer(context.Context, string) error { - f.stops.Add(1) + stops := f.stops.Add(1) + if f.stopErrAt > 0 && stops != f.stopErrAt { + return nil + } return f.stopErr } @@ -185,3 +189,21 @@ func TestIdleControllerWakeReadinessUsesRemainingTimeout(t *testing.T) { assert.Positive(t, remaining) assert.Less(t, remaining, wakeTimeout) } + +func TestIdleControllerRecoversFromPartialStopFailure(t *testing.T) { + lifecycle := &fakeLifecycle{stopErr: errors.New("stop failed"), stopErrAt: 2} + c := NewIdleController(time.Hour, time.Second, []string{"web-1", "web-2"}, lifecycle, func(time.Duration) error { return nil }) + defer c.Close() + c.mu.Lock() + c.lastRequest = time.Now().Add(-2 * time.Hour) + c.mu.Unlock() + + c.trySleep() + + assert.Equal(t, int32(2), lifecycle.stops.Load()) + assert.Equal(t, IdleStateSleeping, c.StateValue()) + require.NoError(t, c.BeginRequest(context.Background())) + c.EndRequest() + assert.Equal(t, int32(2), lifecycle.starts.Load()) + assert.Equal(t, IdleStateActive, c.StateValue()) +} From 105bd1e82084b2fa83ad5eaff2ea0cd86253a88f Mon Sep 17 00:00:00 2001 From: komagata Date: Thu, 23 Jul 2026 13:05:21 +0900 Subject: [PATCH 17/17] Keep Docker negotiation across cancellation --- internal/server/docker_client.go | 34 +++++++++++++++++++++------ internal/server/docker_client_test.go | 31 ++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 7 deletions(-) diff --git a/internal/server/docker_client.go b/internal/server/docker_client.go index 3abda318..fc2c9ff0 100644 --- a/internal/server/docker_client.go +++ b/internal/server/docker_client.go @@ -21,9 +21,10 @@ const ( type DockerClient struct { httpClient *http.Client - versionOnce sync.Once - apiVersion string - versionErr error + versionMu sync.Mutex + versionSet bool + apiVersion string + versionErr error } func NewDockerClient(socketPath string) *DockerClient { @@ -68,10 +69,26 @@ func (c *DockerClient) containerAction(ctx context.Context, name, action string) } func (c *DockerClient) negotiatedVersion(ctx context.Context) (string, error) { - c.versionOnce.Do(func() { - c.apiVersion, c.versionErr = c.queryVersion(ctx) - }) - return c.apiVersion, c.versionErr + c.versionMu.Lock() + defer c.versionMu.Unlock() + if c.versionSet { + return c.apiVersion, c.versionErr + } + + negotiationCtx, cancel := dockerNegotiationContext(ctx) + defer cancel() + version, err := c.queryVersion(negotiationCtx) + if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + c.apiVersion, c.versionErr, c.versionSet = version, err, true + } + return version, err +} + +func dockerNegotiationContext(ctx context.Context) (context.Context, context.CancelFunc) { + if deadline, ok := ctx.Deadline(); ok { + return context.WithDeadline(context.Background(), deadline) + } + return context.WithCancel(context.Background()) } func (c *DockerClient) queryVersion(ctx context.Context) (string, error) { @@ -81,6 +98,9 @@ func (c *DockerClient) queryVersion(ctx context.Context) (string, error) { } resp, err := c.httpClient.Do(req) if err != nil { + if ctx.Err() != nil { + return "", ctx.Err() + } // Some compatible Docker proxies do not expose /version. Preserve the // legacy behavior and let the versioned operation return the useful error. return legacyDockerAPIVersion, nil diff --git a/internal/server/docker_client_test.go b/internal/server/docker_client_test.go index 2dd7f2a5..284da437 100644 --- a/internal/server/docker_client_test.go +++ b/internal/server/docker_client_test.go @@ -100,6 +100,37 @@ func TestDockerClientNegotiatesOnlyOnceWithConcurrentFirstUse(t *testing.T) { assert.Equal(t, int32(1), versionCalls.Load()) } +func TestDockerClientNegotiationSurvivesCallerCancellation(t *testing.T) { + versionStarted := make(chan struct{}) + releaseVersion := make(chan struct{}) + var startedOnce sync.Once + var actionPath string + client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/version" { + startedOnce.Do(func() { close(versionStarted) }) + select { + case <-releaseVersion: + fmt.Fprint(w, `{"ApiVersion":"1.52","MinAPIVersion":"1.44"}`) + case <-r.Context().Done(): + } + return + } + actionPath = r.URL.Path + w.WriteHeader(http.StatusNoContent) + }) + + ctx, cancel := context.WithCancel(context.Background()) + firstDone := make(chan error, 1) + go func() { firstDone <- client.StartContainer(ctx, "web") }() + <-versionStarted + cancel() + close(releaseVersion) + require.Error(t, <-firstDone) + + require.NoError(t, client.StartContainer(context.Background(), "web")) + assert.Equal(t, "/v1.52/containers/web/start", actionPath) +} + func TestDockerClientIncludesBoundedDockerErrorBody(t *testing.T) { client := testDockerClient(t, func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/version" {