fix(controller): handle unassigned canceled completions

controller/lifecycle_test.go

@@ -173,14 +173,15 @@ func TestUntrackedRunnerEventsRetainUnrelatedAndDockerErrors(t *testing.T) {
 			for _, err := range []error{
 				scaler.HandleJobStarted(ctx, &scaleset.JobStarted{RunnerName: test.name}),
 				scaler.HandleJobCompleted(ctx, &scaleset.JobCompleted{RunnerName: test.name}),
+				scaler.HandleJobCompleted(ctx, &scaleset.JobCompleted{RunnerName: test.name, Result: "canceled"}),
 			} {
 				if err == nil || !strings.Contains(err.Error(), test.want) {
 					t.Errorf("untracked runner error was suppressed: %v", err)
 				}
 			}
 			wantInspections := int32(0)
 			if test.status != 0 {
-				wantInspections = 2
+				wantInspections = 3
 			}
 			if inspections.Load() != wantInspections {
 				t.Errorf("inspections=%d, want %d", inspections.Load(), wantInspections)

controller/scaler.go

@@ -87,6 +87,15 @@ func (s *Scaler) HandleJobStarted(ctx context.Context, job *scaleset.JobStarted)
 }
 
 func (s *Scaler) HandleJobCompleted(ctx context.Context, job *scaleset.JobCompleted) error {
+	s.logger.Info("job completion received", "runner", job.RunnerName, "runnerID", job.RunnerID,
+		"result", job.Result, "runnerAssignTime", job.RunnerAssignTime, "jobID", job.JobID,
+		"runnerRequestID", job.RunnerRequestID, "workflowRunID", job.WorkflowRunID)
+	// The pinned SDK documents cancellation of an assignment before a runner acquires it.
+	if job.RunnerName == "" && job.Result == "canceled" && job.RunnerID == 0 && job.RunnerAssignTime.IsZero() &&
+		job.JobID != "" && job.RunnerRequestID > 0 {
+		s.logger.Info("ignored unassigned canceled job completion", "jobID", job.JobID, "runnerRequestID", job.RunnerRequestID)
+		return nil
+	}
 	id, cleanup, ok := s.runners.markDone(job.RunnerName)
 	if !ok {
 		missing, err := s.missingPriorRunner(ctx, job.RunnerName)

controller/unassigned_completion_test.go

@@ -0,0 +1,144 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/actions/scaleset"
+	"github.com/actions/scaleset/listener"
+	dockerclient "github.com/docker/docker/client"
+)
+
+func unassignedCancellation() scaleset.JobCompleted {
+	return scaleset.JobCompleted{Result: "canceled", JobMessageBase: scaleset.JobMessageBase{JobID: "unassigned-job", RunnerRequestID: 7}}
+}
+
+func TestUnassignedCancellationRequiresUnambiguousIdentity(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		edit func(*scaleset.JobCompleted)
+	}{
+		{name: "unassigned cancellation"},
+		{name: "missing result", edit: func(j *scaleset.JobCompleted) { j.Result = "" }},
+		{name: "success", edit: func(j *scaleset.JobCompleted) { j.Result = "success" }},
+		{name: "failure", edit: func(j *scaleset.JobCompleted) { j.Result = "failure" }},
+		{name: "different spelling", edit: func(j *scaleset.JobCompleted) { j.Result = "cancelled" }},
+		{name: "different case", edit: func(j *scaleset.JobCompleted) { j.Result = "Canceled" }},
+		{name: "runner ID", edit: func(j *scaleset.JobCompleted) { j.RunnerID = 1 }},
+		{name: "negative runner ID", edit: func(j *scaleset.JobCompleted) { j.RunnerID = -1 }},
+		{name: "runner assignment", edit: func(j *scaleset.JobCompleted) { j.RunnerAssignTime = time.Unix(1, 0) }},
+		{name: "missing job ID", edit: func(j *scaleset.JobCompleted) { j.JobID = "" }},
+		{name: "missing request ID", edit: func(j *scaleset.JobCompleted) { j.RunnerRequestID = 0 }},
+		{name: "negative request ID", edit: func(j *scaleset.JobCompleted) { j.RunnerRequestID = -1 }},
+		{name: "unrelated name", edit: func(j *scaleset.JobCompleted) { j.RunnerName = "other-runner" }},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			job := unassignedCancellation()
+			if test.edit != nil {
+				test.edit(&job)
+			}
+			scaler := &Scaler{runners: newRunnerState(), logger: slog.New(slog.DiscardHandler), config: Config{FleetInstance: "example"}}
+			scaler.runners.addBusy("ci-fleet-example-0123abcd", "retained")
+			err := scaler.HandleJobCompleted(context.Background(), &job)
+			if (err != nil) != (test.edit != nil) {
+				t.Fatalf("completion error = %v", err)
+			}
+			if count, busy := scaler.runners.counts(); count != 1 || busy != 1 {
+				t.Fatalf("unassigned event changed runners: count=%d busy=%d", count, busy)
+			}
+			if err := scaler.HandleJobStarted(context.Background(), &scaleset.JobStarted{}); err == nil {
+				t.Fatal("blank job start was accepted")
+			}
+		})
+	}
+}
+
+func TestUnknownBlankCompletionRetainsDiagnosticFields(t *testing.T) {
+	var output bytes.Buffer
+	scaler := &Scaler{runners: newRunnerState(), logger: slog.New(slog.NewJSONHandler(&output, nil))}
+	job := scaleset.JobCompleted{Result: "unexpected", RunnerID: 12, JobMessageBase: scaleset.JobMessageBase{
+		JobID: "unknown-job", RunnerRequestID: 13, WorkflowRunID: 14, RunnerAssignTime: time.Unix(1, 0).UTC(),
+	}}
+	if err := scaler.HandleJobCompleted(context.Background(), &job); err == nil {
+		t.Fatal("contradictory blank completion was accepted")
+	}
+	var event map[string]any
+	if err := json.NewDecoder(&output).Decode(&event); err != nil {
+		t.Fatal(err)
+	}
+	for key, want := range map[string]any{"runner": "", "result": "unexpected", "runnerID": float64(12),
+		"jobID": "unknown-job", "runnerRequestID": float64(13), "workflowRunID": float64(14), "runnerAssignTime": "1970-01-01T00:00:01Z"} {
+		if event[key] != want {
+			t.Errorf("diagnostic %s = %v, want %v", key, event[key], want)
+		}
+	}
+}
+
+func TestUnassignedCompletionPreservesTheRestOfTheSDKBatch(t *testing.T) {
+	for _, invalid := range []bool{false, true} {
+		name := "documented cancellation"
+		if invalid {
+			name = "contradictory cancellation"
+		}
+		t.Run(name, func(t *testing.T) {
+			job := unassignedCancellation()
+			if invalid {
+				job.RunnerID = 1
+			}
+			const live = "ci-fleet-example-0123abcd"
+			client := &lifecycleBatchClient{message: &scaleset.RunnerScaleSetMessage{
+				MessageID: 8, Statistics: &scaleset.RunnerScaleSetStatistic{},
+				JobCompletedMessages: []*scaleset.JobCompleted{&job, {RunnerName: live, Result: "canceled"}},
+			}}
+			var removals atomic.Int32
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if !client.acknowledged.Load() {
+					t.Error("Docker access preceded SDK acknowledgment")
+				}
+				switch {
+				case r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/containers/live-id/logs"):
+					w.WriteHeader(http.StatusOK)
+				case r.Method == http.MethodDelete && strings.HasSuffix(r.URL.Path, "/containers/live-id"):
+					removals.Add(1)
+					w.WriteHeader(http.StatusNoContent)
+				default:
+					t.Errorf("unexpected Docker request: %s %s", r.Method, r.URL)
+					w.WriteHeader(http.StatusInternalServerError)
+				}
+			}))
+			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{runners: newRunnerState(), dockerClient: docker, logger: slog.New(slog.DiscardHandler),
+				config: Config{FleetInstance: "example", MaxRunners: 1, StatusFile: t.TempDir() + "/status.json"}}
+			scaler.runners.addBusy(live, "live-id")
+			l, err := listener.New(client, listener.Config{ScaleSetID: 42, MaxRunners: 1})
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = l.Run(context.Background(), scaler)
+			if !client.acknowledged.Load() {
+				t.Fatal("SDK did not acknowledge the batch")
+			}
+			if invalid {
+				if err == nil || !strings.Contains(err.Error(), "unknown runner") || client.polls != 1 || removals.Load() != 0 || scaler.runners.count() != 1 {
+					t.Fatalf("invalid completion was not rejected: err=%v polls=%d removals=%d", err, client.polls, removals.Load())
+				}
+			} else if !errors.Is(err, context.Canceled) || client.polls != 2 || removals.Load() != 1 || scaler.runners.count() != 0 {
+				t.Fatalf("valid cancellation interrupted the batch: err=%v polls=%d removals=%d", err, client.polls, removals.Load())
+			}
+		})
+	}
+}

docs/CONTROLLER-DESIGN.md

@@ -36,6 +36,10 @@ On restart, it recovers only managed runners carrying the same fleet-instance an
 
 The listener ignores late authenticated events for this instance's generated runner names only after Docker confirms their containers are gone; unrelated names and inspection errors still fail.
 
+The pinned SDK also documents canceled scale-set assignments that never reach a runner. The controller ignores such a completion only when `result` is exactly `canceled`, `runnerName` is empty, `runnerId` is zero, `runnerAssignTime` is unset, the job ID is nonempty, and the runner request ID is positive. This conservative exception does not complete or remove any managed runner. Scaling continues to use the scale-set statistics. A blank runner name alone does not prove assignment cancellation.
+
+Before handling a completion, the controller logs its result, runner name and ID, runner assignment time, job ID, runner request ID, and workflow run ID. Unknown or contradictory events and Docker inspection or cleanup errors still fail. The pinned listener acknowledges a batch before invoking callbacks, so a callback failure can stop the rest of that batch; do not assume the failed event will be replayed or that restarting fixes its cause.
+
 ## Security boundary
 
 Docker socket access is host-root-equivalent. This pool is therefore limited to trusted repositories and trusted workflow revisions. Containers provide repeatability and cleanup; they do not make hostile workflow code safe.