ateworker: report ready only after sandbox runtimes are registered

cmd/ateworker-gvisor/main.go

@@ -369,7 +369,7 @@ type AteWorkerService struct {
 	shuttingDown atomic.Bool
 
 	// epoch is the epoch this process registered with. It starts no workload
-	// before registering.
+	// before reporting its sandbox runtimes.
 	epoch register.Epoch
 
 	// cgroupRoot is where the sandbox's cgroup v2 leaves live: the worker pod's

cmd/ateworker-microvm/main.go

@@ -366,7 +366,7 @@ type AteWorkerService struct {
 	shuttingDown atomic.Bool
 
 	// epoch is the epoch this process registered with. It starts no workload
-	// before registering.
+	// before reporting its sandbox runtimes.
 	epoch register.Epoch
 
 	podUID     string

internal/ateworker/register/register.go

@@ -52,14 +52,15 @@ const (
 	resyncInterval       = 5 * time.Minute
 )
 
-// Epoch is the epoch this worker process registered with. A worker process
-// takes no workloads until it has registered, and reports this epoch with each
-// one it starts, so the control plane can tell which process ran it.
+// Epoch is the epoch this worker process registered with, set once the control
+// plane holds its capacity and its sandbox runtimes. A worker process takes no
+// workloads until then, and reports this epoch with each one it starts, so the
+// control plane can tell which process ran it.
 type Epoch struct {
 	v atomic.Int64
 }
 
-// Load returns the registered epoch, or 0 before registration.
+// Load returns the registered epoch, or 0 until sandbox runtimes are reported.
 func (e *Epoch) Load() int64 {
 	return e.v.Load()
 }
@@ -70,22 +71,24 @@ func (e *Epoch) Store(epoch int64) {
 }
 
 // Registered is a readiness check that fails until this worker process has
-// registered. The pod is then not Ready, so the Worker takes no placements
-// while a restarted worker process has yet to register.
+// registered and reported its sandbox runtimes, which is when the control
+// plane can first place on it. The pod is not Ready before then, so the
+// Deployment does not count a worker with no usable runtime as Ready.
 func (e *Epoch) Registered(context.Context) error {
 	if e.Load() == 0 {
-		return errors.New("worker has not registered yet")
+		return errors.New("worker has not reported its sandbox runtimes yet")
 	}
 	return nil
 }
 
 // Require returns the registered epoch for a workload about to start, or
-// Unavailable before registration, so the caller retries once it has.
+// Unavailable until sandbox runtimes are reported, so the caller retries once
+// they are.
 func (e *Epoch) Require() (int64, error) {
 	if epoch := e.Load(); epoch != 0 {
 		return epoch, nil
 	}
-	return 0, apierror.Unavailable("worker has not registered yet: not accepting workloads")
+	return 0, apierror.Unavailable("worker has not reported its sandbox runtimes yet: not accepting workloads")
 }
 
 // FromFiles combines the downward API compute limits with the worker's own
@@ -244,10 +247,10 @@ type supportClient interface {
 	SetWorkerSandboxRuntimes(ctx context.Context, req *ateletpb.SetWorkerSandboxRuntimesRequest) error
 }
 
-// Report registers this worker process with the node-local atelet, stores the
-// epoch it registered with in epoch, then keeps its sandbox runtimes reported
-// until ctx ends: it reads its sandbox runtimes file at startup and again
-// whenever atelet rewrites it.
+// Report registers this worker process with the node-local atelet, then keeps
+// its sandbox runtimes reported until ctx ends: it reads its sandbox runtimes
+// file at startup and again whenever atelet rewrites it. It stores the epoch it
+// registered with in epoch once the first report is accepted.
 //
 // It returns early on a configuration error, which no retry would fix, and
 // when the control plane refuses this process's epoch, which means another
@@ -290,12 +293,21 @@ func run(ctx context.Context, cfg ReportConfig, capacity *ateletpb.WorkerResourc
 	if err != nil {
 		return err
 	}
-	epoch.Store(registered)
+	// Stored once the control plane holds this process's sandbox runtimes as
+	// well as its capacity. A workload that arrives before the response does is
+	// refused with Unavailable and retried, the same as one before registration.
+	report := func(ctx context.Context, req *ateletpb.SetWorkerSandboxRuntimesRequest) error {
+		if err := client.SetWorkerSandboxRuntimes(ctx, req); err != nil {
+			return err
+		}
+		epoch.Store(registered)
+		return nil
+	}
 	r := newReconciler(
 		cfg,
 		registered,
 		func() (*sandboxruntimes.File, error) { return sandboxruntimes.Read(cfg.RuntimesFile) },
-		client.SetWorkerSandboxRuntimes,
+		report,
 	)
 	return r.run(ctx, changes)
 }

internal/ateworker/register/register_test.go

@@ -172,7 +172,8 @@ func TestEpochRequire(t *testing.T) {
 }
 
 // The readiness check holds the pod not Ready until the worker process has
-// registered, so a restarted one takes no placements before it can serve them.
+// reported its sandbox runtimes, so it takes no placements before it can serve
+// them.
 func TestEpochRegistered(t *testing.T) {
 	var e Epoch
 	if err := e.Registered(context.Background()); err == nil {
@@ -270,38 +271,55 @@ func TestRegisterOnce(t *testing.T) {
 	}
 }
 
-func TestRunStoresEpochBeforeReportingRuntimes(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+// run stores the epoch once the control plane has accepted the sandbox
+// runtimes, so a worker whose report has not landed stays not Ready and
+// refuses workloads.
+func TestRunStoresEpochAfterFirstReport(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		reportErr error
+		wantEpoch int64
+	}{
+		{name: "report accepted", wantEpoch: 1},
+		{name: "report failed", reportErr: status.Error(codes.Unavailable, "ateapi is down")},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
 
-	path := filepath.Join(t.TempDir(), nodepath.SandboxRuntimesFileName)
-	if _, err := sandboxruntimes.Write(path, gvisorFile(sandboxruntimes.Attributes{hardware.AttrSandboxBundleHash: "sha256:v1"})); err != nil {
-		t.Fatal(err)
-	}
-	var epoch Epoch
-	client := &runtimesRecorder{fakeSupport: &fakeSupport{}, onReport: func(req *ateletpb.SetWorkerSandboxRuntimesRequest) {
-		if got := epoch.Load(); got != req.GetEpoch() {
-			t.Errorf("epoch stored = %d when runtimes were reported with %d", got, req.GetEpoch())
-		}
-		cancel()
-	}}
-	err := run(ctx, ReportConfig{SandboxClass: "gvisor", RuntimesFile: path, PodIPs: []string{"10.0.0.1"}}, &ateletpb.WorkerResources{Actors: testActors}, client, &epoch, nil)
-	if !errors.Is(err, context.Canceled) {
-		t.Fatalf("run() = %v, want context.Canceled", err)
-	}
-	if epoch.Load() != 1 {
-		t.Errorf("epoch = %d, want 1", epoch.Load())
+			path := filepath.Join(t.TempDir(), nodepath.SandboxRuntimesFileName)
+			if _, err := sandboxruntimes.Write(path, gvisorFile(sandboxruntimes.Attributes{hardware.AttrSandboxBundleHash: "sha256:v1"})); err != nil {
+				t.Fatal(err)
+			}
+			var epoch Epoch
+			client := &runtimesRecorder{fakeSupport: &fakeSupport{}, err: tc.reportErr, onReport: func(*ateletpb.SetWorkerSandboxRuntimesRequest) {
+				if got := epoch.Load(); got != 0 {
+					t.Errorf("epoch = %d while runtimes were being reported, want 0", got)
+				}
+				cancel()
+			}}
+			err := run(ctx, ReportConfig{SandboxClass: "gvisor", RuntimesFile: path, PodIPs: []string{"10.0.0.1"}}, &ateletpb.WorkerResources{Actors: testActors}, client, &epoch, nil)
+			if !errors.Is(err, context.Canceled) {
+				t.Fatalf("run() = %v, want context.Canceled", err)
+			}
+			if got := epoch.Load(); got != tc.wantEpoch {
+				t.Errorf("epoch = %d, want %d", got, tc.wantEpoch)
+			}
+		})
 	}
 }
 
+// runtimesRecorder is a fakeSupport that hands each sandbox runtimes report to
+// onReport and returns err.
 type runtimesRecorder struct {
 	*fakeSupport
 	onReport func(*ateletpb.SetWorkerSandboxRuntimesRequest)
+	err      error
 }
 
 func (r *runtimesRecorder) SetWorkerSandboxRuntimes(_ context.Context, req *ateletpb.SetWorkerSandboxRuntimesRequest) error {
 	r.onReport(req)
-	return nil
+	return r.err
 }
 
 func TestReconcilerSync(t *testing.T) {