From 6595ba8458e43f20c76b43603e00f31d3ce8c448 Mon Sep 17 00:00:00 2001 From: Nick Foster Date: Sat, 17 Oct 2026 13:12:38 -0500 Subject: [PATCH] fix(controller): wait for overlapping runner removal Wait up to 30 seconds for Docker to confirm removal when completion and exit cleanup overlap. Keep unrelated Docker failures visible and cover listener batch continuation, missing containers, cancellation and failed waits. Fixes #143 --- controller/cleanup_test.go | 211 +++++++++++++++++++++++++++++++++++++ controller/scaler.go | 18 +++- docs/ARCHITECTURE.md | 6 ++ 3 files changed, 234 insertions(+), 1 deletion(-) create mode 100644 controller/cleanup_test.go diff --git a/controller/cleanup_test.go b/controller/cleanup_test.go new file mode 100644 index 00000000..36c6915a --- /dev/null +++ b/controller/cleanup_test.go @@ -0,0 +1,211 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/actions/scaleset" + "github.com/actions/scaleset/listener" + dockerclient "github.com/docker/docker/client" +) + +func TestOverlappingRunnerCleanupPreservesTheListenerBatch(t *testing.T) { + const runner = "ci-fleet-example-0123abcd" + const other = "ci-fleet-example-89abcdef" + removing := make(chan struct{}) + finishRemoval := make(chan struct{}) + removed := make(chan struct{}) + var finish sync.Once + var removals, otherRemovals atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + path := strings.TrimPrefix(r.URL.Path, "/v1.48") + switch { + case path == "/containers/runner-id/wait": + switch r.URL.Query().Get("condition") { + case "not-running": + json.NewEncoder(w).Encode(map[string]int{"StatusCode": 0}) + case "removed": + // Release the watcher's DELETE only once completion waits for removal. + finish.Do(func() { close(finishRemoval) }) + select { + case <-removed: + json.NewEncoder(w).Encode(map[string]int{"StatusCode": 0}) + case <-r.Context().Done(): + } + default: + t.Errorf("unexpected wait condition: %s", r.URL) + w.WriteHeader(http.StatusBadRequest) + } + case path == "/containers/runner-id/json": + json.NewEncoder(w).Encode(map[string]any{"State": map[string]bool{"Running": false}}) + case strings.HasSuffix(path, "/logs"): + w.WriteHeader(http.StatusOK) + case path == "/containers/runner-id" && r.Method == http.MethodDelete: + if removals.Add(1) == 1 { + if r.URL.Query().Get("force") == "1" { + t.Error("exit watcher forced removal") + } + close(removing) + select { + case <-finishRemoval: + close(removed) + w.WriteHeader(http.StatusNoContent) + case <-r.Context().Done(): + } + } else { + if r.URL.Query().Get("force") != "1" { + t.Error("job completion did not force removal") + } + w.WriteHeader(http.StatusConflict) + json.NewEncoder(w).Encode(map[string]string{"message": "removal of container runner-id is already in progress"}) + } + case path == "/containers/other-id" && r.Method == http.MethodDelete: + otherRemovals.Add(1) + w.WriteHeader(http.StatusNoContent) + default: + t.Errorf("unexpected Docker request %s %s", r.Method, r.URL) + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + defer finish.Do(func() { close(finishRemoval) }) + docker, err := dockerclient.NewClientWithOpts(dockerclient.WithHost("tcp://"+server.Listener.Addr().String()), dockerclient.WithHTTPClient(server.Client()), dockerclient.WithVersion("1.48")) + if err != nil { + t.Fatal(err) + } + defer docker.Close() + scaler := &Scaler{runners: newRunnerState(), dockerClient: docker, logger: slog.New(slog.DiscardHandler), + config: Config{MaxRunners: 2, StatusFile: t.TempDir() + "/status.json"}} + scaler.runners.addBusy(runner, "runner-id") + scaler.runners.addBusy(other, "other-id") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + watchDone := make(chan struct{}) + go func() { + scaler.watchRunner(ctx, runner, "runner-id") + close(watchDone) + }() + select { + case <-removing: + case <-ctx.Done(): + t.Fatal("watcher did not begin removal") + } + client := &lifecycleBatchClient{message: &scaleset.RunnerScaleSetMessage{ + MessageID: 7, Statistics: &scaleset.RunnerScaleSetStatistic{}, + JobCompletedMessages: []*scaleset.JobCompleted{{RunnerName: runner}, {RunnerName: other}}, + }} + l, err := listener.New(client, listener.Config{ScaleSetID: 42, MaxRunners: 2}) + if err != nil { + t.Fatal(err) + } + if err := l.Run(ctx, scaler); !errors.Is(err, context.Canceled) || ctx.Err() != nil { + t.Fatalf("overlapping cleanup interrupted the listener batch: %v", err) + } + select { + case <-watchDone: + case <-ctx.Done(): + t.Fatal("exit watcher did not finish") + } + if !client.acknowledged.Load() || client.polls != 2 || removals.Load() != 2 || otherRemovals.Load() != 1 { + t.Fatalf("batch did not finish: acknowledged=%v polls=%d removals=%d other removals=%d", client.acknowledged.Load(), client.polls, removals.Load(), otherRemovals.Load()) + } + if scaler.runners.contains(runner, "runner-id") || scaler.runners.count() != 0 { + t.Fatal("completed runners retained lifecycle state or capacity") + } + select { + case <-removed: + default: + t.Fatal("cleanup succeeded before container removal") + } +} + +func TestRunnerRemovalResults(t *testing.T) { + const inProgress = "removal of container runner-id is already in progress" + for _, test := range []struct { + name, message, waitBody, want string + removeStatus, waitStatus int + cancelWait bool + }{ + {name: "removed", removeStatus: http.StatusNoContent}, + {name: "already absent", removeStatus: http.StatusNotFound}, + {name: "daemon error", removeStatus: http.StatusInternalServerError, message: "daemon failure", want: "daemon failure"}, + {name: "permission denied", removeStatus: http.StatusForbidden, message: "permission denied", want: "permission denied"}, + {name: "unrelated conflict", removeStatus: http.StatusConflict, message: "container is running", want: "container is running"}, + {name: "another container", removeStatus: http.StatusConflict, message: "removal of container other-id is already in progress", want: "other-id"}, + {name: "wrong error class", removeStatus: http.StatusInternalServerError, message: inProgress, want: inProgress}, + {name: "removed during wait", removeStatus: http.StatusConflict, message: inProgress, waitStatus: http.StatusOK, waitBody: `{"StatusCode":0}`}, + {name: "absent before wait", removeStatus: http.StatusConflict, message: inProgress, waitStatus: http.StatusNotFound}, + {name: "wait daemon error", removeStatus: http.StatusConflict, message: inProgress, waitStatus: http.StatusInternalServerError, want: "wait failure"}, + {name: "wait result error", removeStatus: http.StatusConflict, message: inProgress, waitStatus: http.StatusOK, waitBody: `{"Error":{"Message":"removal failed"}}`, want: "removal failed"}, + {name: "incomplete wait response", removeStatus: http.StatusConflict, message: inProgress, waitStatus: http.StatusOK, waitBody: `{"StatusCode":`, want: "unexpected EOF"}, + {name: "deadline", removeStatus: http.StatusConflict, message: inProgress, want: "context deadline exceeded"}, + {name: "cancellation", removeStatus: http.StatusConflict, message: inProgress, want: "context canceled", cancelWait: true}, + } { + t.Run(test.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond) + defer cancel() + var waits atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasSuffix(r.URL.Path, "/logs"): + w.WriteHeader(http.StatusOK) + case r.Method == http.MethodDelete: + w.WriteHeader(test.removeStatus) + if test.removeStatus != http.StatusNoContent { + json.NewEncoder(w).Encode(map[string]string{"message": test.message}) + } + case strings.HasSuffix(r.URL.Path, "/wait"): + waits.Add(1) + if r.URL.Query().Get("condition") != "removed" { + t.Error("wait did not require container removal") + } + if test.waitStatus == 0 { + if test.cancelWait { + cancel() + } + <-r.Context().Done() + return + } + w.WriteHeader(test.waitStatus) + if test.waitStatus != http.StatusOK { + json.NewEncoder(w).Encode(map[string]string{"message": "wait failure"}) + } else { + w.Write([]byte(test.waitBody)) + } + default: + t.Errorf("unexpected Docker request %s %s", r.Method, r.URL) + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + docker, err := dockerclient.NewClientWithOpts(dockerclient.WithHost("tcp://"+server.Listener.Addr().String()), dockerclient.WithHTTPClient(server.Client()), dockerclient.WithVersion("1.48")) + if err != nil { + t.Fatal(err) + } + defer docker.Close() + scaler := &Scaler{dockerClient: docker, logger: slog.New(slog.DiscardHandler)} + err = scaler.logAndRemove(ctx, "runner", "runner-id", true) + if test.want == "" && err != nil || test.want != "" && (err == nil || !strings.Contains(err.Error(), test.want)) { + t.Fatalf("cleanup error = %v, want %q", err, test.want) + } + wantWaits := int32(0) + if test.removeStatus == http.StatusConflict && test.message == inProgress { + wantWaits = 1 + } + if waits.Load() != wantWaits { + t.Fatalf("removal waits = %d, want %d", waits.Load(), wantWaits) + } + }) + } +} diff --git a/controller/scaler.go b/controller/scaler.go index f5758895..da1bc612 100644 --- a/controller/scaler.go +++ b/controller/scaler.go @@ -290,7 +290,23 @@ func (s *Scaler) logAndRemove(ctx context.Context, name, id string, force bool) s.logger.Warn("could not collect runner logs", "runner", name, "error", err) } if err := s.dockerClient.ContainerRemove(ctx, id, container.RemoveOptions{Force: force, RemoveVolumes: true}); err != nil && !errdefs.IsNotFound(err) { - return fmt.Errorf("remove runner %s: %w", name, err) + if !errdefs.IsConflict(err) || !strings.Contains(err.Error(), "removal of container "+id+" is already in progress") { + return fmt.Errorf("remove runner %s: %w", name, err) + } + // Completion and exit cleanup can overlap; confirm the other removal finishes. + waitCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + removed, errors := s.dockerClient.ContainerWait(waitCtx, id, container.WaitConditionRemoved) + select { + case result := <-removed: + if result.Error != nil { + return fmt.Errorf("wait for runner %s removal: %s", name, result.Error.Message) + } + case err := <-errors: + if !errdefs.IsNotFound(err) { + return fmt.Errorf("wait for runner %s removal: %w", name, err) + } + } } return nil } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index b296e361..3110a968 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -146,6 +146,12 @@ flowchart TD Host-wide pruning must not run as an uncoordinated per-job operation. Hard cancellation and host failure may bypass project traps, so resources must remain identifiable by run and age. +Job completion and the runner exit watcher can request removal of the same +container concurrently. If Docker reports that removal is already in progress, +the controller waits up to 30 seconds for that container's removal. Confirmed +removal or an already absent container lets processing continue. Other Docker +errors, cancellation and an expired wait remain cleanup failures. + The existing host cleanup timer also reclaims empty networks at least ten minutes old whose allocations lie entirely inside rendered Docker default address pools. This includes