diff --git a/cmd/ateapi/internal/controlapi/functionaltest/actor_volume_detach_test.go b/cmd/ateapi/internal/controlapi/functionaltest/actor_volume_detach_test.go new file mode 100644 index 0000000000..714568ea78 --- /dev/null +++ b/cmd/ateapi/internal/controlapi/functionaltest/actor_volume_detach_test.go @@ -0,0 +1,315 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package functionaltest + +import ( + "context" + "slices" + "testing" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/internal/volume" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// calls returns copies of the nodes the plugin attached to and detached from, +// in call order. +func (a *attachFailVolumePlugin) calls() (attached, detached []string) { + a.mu.Lock() + defer a.mu.Unlock() + return slices.Clone(a.attachedNodes), slices.Clone(a.detachedNodes) +} + +// assertAssignedNode checks the actor's persisted assigned_node in the store +// against want. +func assertAssignedNode(t *testing.T, tc *testContext, name, step, want string) { + t.Helper() + actor, err := tc.persistence.GetActor(context.Background(), resources.ActorRef{Atespace: testAtespace, Name: name}) + if err != nil { + t.Fatalf("GetActor(%s) from store: %v", name, err) + } + if got := actor.GetStatus().GetAssignedNode(); got != want { + t.Errorf("persisted assigned_node %s = %q, want %q", step, got, want) + } +} + +// assertDetachedNodes checks the nodes the plugin detached from, in call +// order. An empty want means no detach. +func assertDetachedNodes(t *testing.T, plugin *attachFailVolumePlugin, step string, want ...string) { + t.Helper() + _, detached := plugin.calls() + if diff := cmp.Diff(want, detached, cmpopts.EquateEmpty()); diff != "" { + t.Errorf("detached nodes %s (-want +got):\n%s", step, diff) + } +} + +// createSingleVolumeActor creates a template mounting one external volume, +// worker-1 on node1, and a SUSPENDED actor named name from the template. It +// returns worker-1's Worker name. +func createSingleVolumeActor(t *testing.T, tc *testContext, ns, name string) string { + t.Helper() + volumes := []*ateapipb.Volume{{ + Name: "vol1", + ExternalVolumeTemplate: &ateapipb.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: "10Gi", + }, + }} + mounts := []*ateapipb.VolumeMount{{Name: "vol1", MountPath: "/mnt/vol1"}} + createTemplateWithVolumes(t, tc, ns, volumes, mounts) + workerName := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + if _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: name}, + ActorTemplate: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "tmpl1"}, + }}); err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + return workerName +} + +// resumeToRunningOnNode1 resumes the actor and checks that its volume was +// published to node1 and that the store records node1 as its assigned_node. +func resumeToRunningOnNode1(t *testing.T, tc *testContext, plugin *attachFailVolumePlugin, name string) { + t.Helper() + if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, + }); err != nil { + t.Fatalf("ResumeActor failed: %v", err) + } + if attached, _ := plugin.calls(); !cmp.Equal(attached, []string{"node1"}) { + t.Fatalf("attached nodes after resume = %v, want [node1]", attached) + } + assertAssignedNode(t, tc, name, "after resume", "node1") +} + +// assertCrashedKeepingAssignedNode checks that the actor is CRASHED with no +// worker assignment, and that its status still records node as assigned_node. +func assertCrashedKeepingAssignedNode(t *testing.T, tc *testContext, name, node string) { + t.Helper() + actor, err := tc.persistence.GetActor(context.Background(), resources.ActorRef{Atespace: testAtespace, Name: name}) + if err != nil { + t.Fatalf("GetActor(%s) from store: %v", name, err) + } + if got := actor.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_CRASHED { + t.Fatalf("state = %v, want CRASHED", got) + } + if got := actor.GetStatus().GetWorkerAssignment(); got != nil { + t.Fatalf("worker assignment = %v, want nil after the crash", got) + } + assertAssignedNode(t, tc, name, "after the crash", node) +} + +// TestDeleteActor_CrashedWorkerGone_DetachesAssignedNode verifies that +// deleting an actor that crashed because its worker went away unpublishes its +// volume from the node recorded in assigned_node, although the crash cleared +// the worker assignment and the worker no longer exists. +func TestDeleteActor_CrashedWorkerGone_DetachesAssignedNode(t *testing.T) { + ns := namespaceForTest("ns-del-crashed-detach") + plugin := &attachFailVolumePlugin{} + tc := setupTestWithVolumePlugins(t, ns, map[string]volume.VolumePluginControlPlane{ + "substrate.io/mock": plugin, + }) + defer tc.cleanup() + + const name = "crashed-delete-actor" + createSingleVolumeActor(t, tc, ns, name) + resumeToRunningOnNode1(t, tc, plugin, name) + + // Deregistering the worker crashes the actor it hosts. + deleteWorkerPod(t, tc, ns, "worker-1") + assertCrashedKeepingAssignedNode(t, tc, name, "node1") + assertDetachedNodes(t, plugin, "after the crash") + + // CRASHED is deletable without AnyState. + if _, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, + }); err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + assertDetachedNodes(t, plugin, "after delete", "node1") + plugin.mu.Lock() + if len(plugin.deleted) != 1 { + t.Errorf("deleted volumes = %v, want exactly one", plugin.deleted) + } + plugin.mu.Unlock() +} + +// TestResumeActor_RestoreCrash_DeleteDetaches verifies that when Restore fails +// with an error a retry cannot clear after the volume was published, the +// crashed actor keeps assigned_node on record and a later delete unpublishes +// the volume from it. +func TestResumeActor_RestoreCrash_DeleteDetaches(t *testing.T) { + ns := namespaceForTest("ns-restore-crash-detach") + plugin := &attachFailVolumePlugin{} + tc := setupTestWithVolumePlugins(t, ns, map[string]volume.VolumePluginControlPlane{ + "substrate.io/mock": plugin, + }) + defer tc.cleanup() + + const name = "restore-crash-actor" + createSingleVolumeActor(t, tc, ns, name) + tc.fakeAtelet.FailRestore = status.Error(codes.Internal, "injected restore failure") + + _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, + }) + if got := status.Code(err); got != codes.Internal { + t.Fatalf("ResumeActor code = %v, want %v (err: %v)", got, codes.Internal, err) + } + if attached, _ := plugin.calls(); !cmp.Equal(attached, []string{"node1"}) { + t.Fatalf("attached nodes = %v, want [node1]: the volume must be published before Restore fails", attached) + } + assertCrashedKeepingAssignedNode(t, tc, name, "node1") + assertDetachedNodes(t, plugin, "after the crash") + + if _, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, + }); err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + assertDetachedNodes(t, plugin, "after delete", "node1") +} + +// TestSuspendActor_WorkerRowGone_StillDetaches verifies that a suspend +// unpublishes the volume from the node recorded on the actor even when the +// worker row the assignment names is already gone, and clears assigned_node +// afterwards. +func TestSuspendActor_WorkerRowGone_StillDetaches(t *testing.T) { + ns := namespaceForTest("ns-suspend-worker-gone") + plugin := &attachFailVolumePlugin{} + tc := setupTestWithVolumePlugins(t, ns, map[string]volume.VolumePluginControlPlane{ + "substrate.io/mock": plugin, + }) + defer tc.cleanup() + + const name = "worker-gone-actor" + workerName := createSingleVolumeActor(t, tc, ns, name) + resumeToRunningOnNode1(t, tc, plugin, name) + + // Remove the worker row directly, bypassing the DeleteWorker workflow, so + // the actor stays RUNNING with an assignment to a worker that no longer + // exists. + if _, err := tc.persistence.DeleteWorker(context.Background(), workerName, store.DeletePreconditions{}); err != nil { + t.Fatalf("DeleteWorker(%s) from store: %v", workerName, err) + } + + suspended, err := tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, + }) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + if got := suspended.GetActor().GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { + t.Errorf("state = %v, want SUSPENDED", got) + } + assertDetachedNodes(t, plugin, "after suspend", "node1") + assertAssignedNode(t, tc, name, "after suspend", "") +} + +// TestRevertActor_Crashed_DetachesAssignedNode verifies that reverting an +// actor that crashed because its worker went away unpublishes its volume from +// the node recorded in assigned_node, clears assigned_node, and leaves the +// actor SUSPENDED. +func TestRevertActor_Crashed_DetachesAssignedNode(t *testing.T) { + ns := namespaceForTest("ns-revert-crashed-detach") + plugin := &attachFailVolumePlugin{} + tc := setupTestWithVolumePlugins(t, ns, map[string]volume.VolumePluginControlPlane{ + "substrate.io/mock": plugin, + }) + defer tc.cleanup() + + const name = "crashed-revert-actor" + createSingleVolumeActor(t, tc, ns, name) + resumeToRunningOnNode1(t, tc, plugin, name) + + deleteWorkerPod(t, tc, ns, "worker-1") + assertCrashedKeepingAssignedNode(t, tc, name, "node1") + tc.fakeAtelet.Lock.Lock() + tc.fakeAtelet.TerminateCalled = false + tc.fakeAtelet.Lock.Unlock() + + reverted, err := tc.client.RevertActor(context.Background(), &ateapipb.RevertActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, + }) + if err != nil { + t.Fatalf("RevertActor failed: %v", err) + } + if got := reverted.GetActor().GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { + t.Errorf("state = %v, want SUSPENDED", got) + } + assertDetachedNodes(t, plugin, "after revert", "node1") + assertAssignedNode(t, tc, name, "after revert", "") + tc.fakeAtelet.Lock.Lock() + if !tc.fakeAtelet.TerminateCalled { + t.Errorf("expected Terminate call to clean up assigned node") + } else if gotUID := tc.fakeAtelet.TerminateRequest.GetWorkerPodUid(); gotUID != "" { + t.Errorf("TerminateRequest.WorkerPodUid = %q, want empty", gotUID) + } + tc.fakeAtelet.Lock.Unlock() +} + +// TestRevertActor_WorkerNoLongerHosts_Detaches verifies that reverting a +// RUNNING actor whose worker no longer hosts it skips the ateom workload +// termination (WorkerPodUid == "") but still unpublishes the volume from the +// node recorded on the actor and clears assigned_node. +func TestRevertActor_WorkerNoLongerHosts_Detaches(t *testing.T) { + ns := namespaceForTest("ns-revert-unhosted-detach") + plugin := &attachFailVolumePlugin{} + tc := setupTestWithVolumePlugins(t, ns, map[string]volume.VolumePluginControlPlane{ + "substrate.io/mock": plugin, + }) + defer tc.cleanup() + + const name = "unhosted-revert-actor" + workerName := createSingleVolumeActor(t, tc, ns, name) + resumeToRunningOnNode1(t, tc, plugin, name) + + actor, err := tc.persistence.GetActor(context.Background(), resources.ActorRef{Atespace: testAtespace, Name: name}) + if err != nil { + t.Fatalf("GetActor(%s) from store: %v", name, err) + } + // Drop the worker's assignment row only; the actor still names the worker. + if _, err := tc.persistence.ReleaseActorFromWorker(context.Background(), workerName, actor.GetMetadata().GetUid()); err != nil { + t.Fatalf("ReleaseActorFromWorker failed: %v", err) + } + tc.fakeAtelet.Lock.Lock() + tc.fakeAtelet.TerminateCalled = false + tc.fakeAtelet.Lock.Unlock() + + reverted, err := tc.client.RevertActor(context.Background(), &ateapipb.RevertActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, + }) + if err != nil { + t.Fatalf("RevertActor failed: %v", err) + } + if got := reverted.GetActor().GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { + t.Errorf("state = %v, want SUSPENDED", got) + } + assertDetachedNodes(t, plugin, "after revert", "node1") + assertAssignedNode(t, tc, name, "after revert", "") + tc.fakeAtelet.Lock.Lock() + if !tc.fakeAtelet.TerminateCalled { + t.Errorf("expected Terminate call to clean up assigned node") + } else if gotUID := tc.fakeAtelet.TerminateRequest.GetWorkerPodUid(); gotUID != "" { + t.Errorf("TerminateRequest.WorkerPodUid = %q, want empty", gotUID) + } + tc.fakeAtelet.Lock.Unlock() +} diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index b93a4c16ac..09ececcdba 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -523,11 +523,16 @@ func (w *ActorWorkflow) assignWorkerAttempt(ctx context.Context, actorRef resour constraints.RequiredRuntime = nil } + ref := &ateapipb.ObjectRef{ + Atespace: actor.GetMetadata().GetAtespace(), + Name: actor.GetMetadata().GetName(), + } + if assignedWorker.GetNodeName() == "" && len(getMountedActorVolumes(ctx, ref, actor.GetStatus().GetExternalVolumes(), actorTemplate)) > 0 { + return nil, nil, apierror.FailedPrecondition("worker %s has no node name to attach the actor's volumes to", assignedWorker.GetMetadata().GetName()) + } + assignment := &ateapipb.ActorAssignment{ - Actor: &ateapipb.ObjectRef{ - Atespace: actor.GetMetadata().GetAtespace(), - Name: actor.GetMetadata().GetName(), - }, + Actor: ref, ActorUid: actor.GetMetadata().GetUid(), // Record what this claim reserves so release returns the same amount. Resources: admittedResources(constraints), @@ -715,8 +720,9 @@ func (w *ActorWorkflow) appliesWithFallback(worker *ateapipb.Worker, constraints // ensureVolumesAttached attaches the actor's mounted external volumes to the // assigned worker's node and returns each driver's attachment metadata by -// volume name, for the Restore call that follows. Attachment is idempotent, so -// a re-entered workflow safely runs it again. +// volume name, for the Restore call that follows. Every volume must be +// provisioned with a storage volume ID; otherwise nothing is attached. +// Attachment is idempotent, so a re-entered workflow safely runs it again. // TODO replace re-execution with a proper check on the volumes' attach state. func (w *ActorWorkflow) ensureVolumesAttached(ctx context.Context, actor *ateapipb.Actor, worker *ateapipb.Worker, actorTemplate *ateapipb.ActorTemplate) (_ map[string]map[string]string, err error) { ctx, done := stepSpan(ctx, "AttachVolumes") @@ -728,8 +734,14 @@ func (w *ActorWorkflow) ensureVolumesAttached(ctx context.Context, actor *ateapi } ref := &ateapipb.ObjectRef{Atespace: actor.GetMetadata().GetAtespace(), Name: actor.GetMetadata().GetName()} + volumes := getMountedActorVolumes(ctx, ref, actor.GetStatus().GetExternalVolumes(), actorTemplate) + // Check every volume before attaching any, so a bad one cannot leave the + // ones before it published. + if err := checkVolumesAttachable(volumes); err != nil { + return nil, err + } volumePublishContexts := make(map[string]map[string]string) - for _, vol := range getMountedActorVolumes(ctx, ref, actor.GetStatus().GetExternalVolumes(), actorTemplate) { + for _, vol := range volumes { slog.InfoContext(ctx, "Attaching volume to node", slog.String("volume_id", vol.GetStorageVolumeId()), slog.String("node", node)) plugin, err := w.pluginRegistry.GetPlugin(ctx, vol.GetVolumeType()) if err != nil { @@ -746,6 +758,17 @@ func (w *ActorWorkflow) ensureVolumesAttached(ctx context.Context, actor *ateapi return volumePublishContexts, nil } +// checkVolumesAttachable returns FailedPrecondition unless every volume has a +// storage volume ID. +func checkVolumesAttachable(volumes []*ateapipb.ExternalVolume) error { + for _, vol := range volumes { + if vol.GetStorageVolumeId() == "" { + return apierror.FailedPrecondition("volume %q has no storage volume ID to attach", vol.GetName()) + } + } + return nil +} + // ensureAteletRestored brings the workload up on the assigned worker: // restoring the actor's local snapshot when one exists, else its external // snapshot, else cold-booting from the template spec. This is the diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 3b7ad34f3c..2891a1bc92 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -19,6 +19,7 @@ import ( "errors" "fmt" "net" + "strings" "sync" "testing" "time" @@ -1660,9 +1661,11 @@ func TestResumeActor_AteletWireRequest(t *testing.T) { type publishContextVolumePlugin struct { volume.VolumePluginControlPlane publishContext map[string]string + attachCalls []volume.AttachVolumeRequest } func (p *publishContextVolumePlugin) AttachVolume(ctx context.Context, req volume.AttachVolumeRequest) (volume.AttachVolumeResponse, error) { + p.attachCalls = append(p.attachCalls, req) return volume.AttachVolumeResponse{PublishContext: p.publishContext}, nil } @@ -2512,6 +2515,154 @@ func TestRequiredSnapshotRuntime(t *testing.T) { actor := &ateapipb.Actor{Status: &ateapipb.ActorStatus{Snapshots: tc.snapshots}} if diff := cmp.Diff(tc.want, requiredSnapshotRuntime(actor, tmpl), protocmp.Transform()); diff != "" { t.Errorf("requiredSnapshotRuntime mismatch (-want +got):\n%s", diff) + + } + }) + } +} + +// mockVolume returns a volume served by the "mock" plugin. +func mockVolume(name, storageID string) *ateapipb.ExternalVolume { + return &ateapipb.ExternalVolume{Name: name, StorageVolumeId: storageID, VolumeType: "mock"} +} + +// volumeTemplate returns a gvisor template that declares the volumes named in +// declared and mounts those named in mounted. +func volumeTemplate(declared []string, mounted ...string) *ateapipb.ActorTemplate { + tmpl := &ateapipb.ActorTemplate{ + SandboxConfig: &ateapipb.SandboxConfig{SandboxClass: "gvisor"}, + Containers: []*ateapipb.Container{{Name: "main", Image: "img"}}, + } + for _, name := range declared { + tmpl.Volumes = append(tmpl.Volumes, &ateapipb.Volume{Name: name, ExternalVolumeTemplate: &ateapipb.ExternalVolumeTemplate{StorageClassName: "sc"}}) + } + for _, name := range mounted { + tmpl.Containers[0].VolumeMounts = append(tmpl.Containers[0].VolumeMounts, &ateapipb.VolumeMount{Name: name, MountPath: "/" + name}) + } + return tmpl +} + +// seedVolumeAssignFixture stores one free gvisor worker on node and a +// SUSPENDED actor holding volumes, and returns the stored actor plus a +// started worker cache. +func seedVolumeAssignFixture(t *testing.T, ctx context.Context, persistence store.Interface, node string, volumes ...*ateapipb.ExternalVolume) (*ateapipb.Actor, *workercache.Cache) { + t.Helper() + if _, err := persistence.CreateWorker(ctx, &ateapipb.Worker{ + Metadata: &ateapipb.ResourceMetadata{Name: testWorkerUID("pod-1")}, + WorkerNamespace: "worker-ns", + WorkerPool: "pool", + WorkerPod: "pod-1", + WorkerPodUid: testWorkerUID("pod-1"), + NodeName: node, + SandboxClass: "gvisor", + Status: &ateapipb.WorkerStatus{State: ateapipb.WorkerState_WORKER_STATE_ACTIVE, Capacity: &ateapipb.WorkerResources{Actors: 1}}, + }); err != nil { + t.Fatalf("CreateWorker: %v", err) + } + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "id1"}, + Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ExternalVolumes: volumes}, + }) + actor, err := persistence.GetActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}) + if err != nil { + t.Fatalf("GetActor: %v", err) + } + cacheCtx, cancel := context.WithCancel(ctx) + t.Cleanup(cancel) + wc := workercache.New(persistence, time.Minute) + if err := wc.Start(cacheCtx); err != nil { + t.Fatalf("workercache.Start: %v", err) + } + return actor, wc +} + +// newVolumeAssignWorkflow returns a workflow that schedules from wc and +// serves "mock" volumes with plugin. +func newVolumeAssignWorkflow(st store.Interface, wc *workercache.Cache, plugin volume.VolumePluginControlPlane) *ActorWorkflow { + return &ActorWorkflow{ + store: st, + workerCache: wc, + scheduler: scheduling.New(wc), + pluginRegistry: &mockPluginRegistry{plugins: map[string]volume.VolumePluginControlPlane{"mock": plugin}}, + } +} + +// TestResume_ChecksMountedVolumesBeforeAttach runs the assignment write and +// the attach step together: the assignment write records AssignedNode on the +// actor, and ensureVolumesAttached publishes only mounted volumes to it. A +// selected volume without a storage volume ID fails the attach step with +// FailedPrecondition before any publish call, even when a valid volume +// precedes it. +func TestResume_ChecksMountedVolumesBeforeAttach(t *testing.T) { + tests := []struct { + name string + // last is the second mounted volume, placed after every other volume. + last *ateapipb.ExternalVolume + // wantErr is a substring of the expected FailedPrecondition; empty + // means the attach succeeds. + wantErr string + }{{ + name: "all provisioned", + last: mockVolume("b", "id-b"), + }, { + name: "mounted volume without storage ID", + last: mockVolume("b", ""), + wantErr: `volume "b" has no storage volume ID`, + }} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + persistence := newTestPersistence(t) + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + actor, wc := seedVolumeAssignFixture(t, ctx, persistence, "node-1", + mockVolume("a", "id-a"), + mockVolume("u", "id-u"), + mockVolume("o", "id-o"), + tt.last, + ) + plugin := &publishContextVolumePlugin{} + w := newVolumeAssignWorkflow(persistence, wc, plugin) + // u is declared but not mounted; o is not declared at all. + tmpl := volumeTemplate([]string{"a", "u", "b"}, "a", "b") + + assigned, worker, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl) + if err != nil { + t.Fatalf("assignWorkerAttempt: %v", err) + } + if got := assigned.GetStatus().GetAssignedNode(); got != "node-1" { + t.Errorf("assigned actor AssignedNode = %q, want node-1", got) + } + _, attachErr := w.ensureVolumesAttached(ctx, assigned, worker, tmpl) + + nameByID := make(map[string]string) + for _, vol := range assigned.GetStatus().GetExternalVolumes() { + nameByID[vol.GetStorageVolumeId()] = vol.GetName() + } + + if tt.wantErr != "" { + if got := apierror.Code(attachErr); got != codes.FailedPrecondition { + t.Fatalf("ensureVolumesAttached error = %v (code %v), want FailedPrecondition", attachErr, got) + } + if !strings.Contains(attachErr.Error(), tt.wantErr) { + t.Errorf("ensureVolumesAttached error = %q, want it to contain %q", attachErr, tt.wantErr) + } + if len(plugin.attachCalls) != 0 { + t.Errorf("attach calls = %v, want none", plugin.attachCalls) + } + return + } + if attachErr != nil { + t.Fatalf("ensureVolumesAttached: %v", attachErr) + } + var published []string + for _, call := range plugin.attachCalls { + if call.Node != "node-1" { + t.Errorf("attach of %q went to node %q, want node-1", call.VolumeID, call.Node) + } + published = append(published, nameByID[call.VolumeID]) + } + if diff := cmp.Diff([]string{"a", "b"}, published); diff != "" { + t.Errorf("published volumes mismatch (-want +got):\n%s", diff) } }) } @@ -2565,3 +2716,66 @@ func TestCanRelaxRuntime(t *testing.T) { }) } } + +// TestAssignWorkerAttempt_WorkerWithoutNodeRejectsVolumes verifies a worker +// with no node name is not claimed for an actor with volumes to attach: the +// attempt fails with FailedPrecondition before the bind, leaving no worker +// claim and no actor write. With no volume to attach the worker is still +// assigned. +func TestAssignWorkerAttempt_WorkerWithoutNodeRejectsVolumes(t *testing.T) { + tests := []struct { + name string + mounted []string + wantErr bool + }{ + {name: "mounted volume", mounted: []string{"vol"}, wantErr: true}, + {name: "unmounted volume", mounted: nil, wantErr: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + persistence := newTestPersistence(t) + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + actor, wc := seedVolumeAssignFixture(t, ctx, persistence, "", + mockVolume("vol", "storage-vol"), + ) + plugin := &publishContextVolumePlugin{} + w := newVolumeAssignWorkflow(persistence, wc, plugin) + tmpl := volumeTemplate([]string{"vol"}, tt.mounted...) + + _, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl) + stored, gerr := persistence.GetActor(ctx, actorRef) + if gerr != nil { + t.Fatalf("GetActor: %v", gerr) + } + claim := firstAssignment(t, persistence, testWorkerUID("pod-1")) + if !tt.wantErr { + if err != nil { + t.Fatalf("assignWorkerAttempt: %v", err) + } + if claim == nil { + t.Error("worker claim = nil, want the actor claimed") + } + if got := stored.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_RESUMING { + t.Errorf("stored state = %v, want RESUMING", got) + } + return + } + if got := apierror.Code(err); got != codes.FailedPrecondition { + t.Fatalf("assignWorkerAttempt error = %v (code %v), want FailedPrecondition", err, got) + } + if want := "has no node name"; !strings.Contains(err.Error(), want) { + t.Errorf("assignWorkerAttempt error = %q, want it to contain %q", err, want) + } + if claim != nil { + t.Errorf("worker claim = %v, want none", claim) + } + if diff := cmp.Diff(actor, stored, protocmp.Transform()); diff != "" { + t.Errorf("assignment wrote to the stored actor (-before +after):\n%s", diff) + } + if len(plugin.attachCalls) != 0 { + t.Errorf("attach calls = %v, want none", plugin.attachCalls) + } + }) + } +} diff --git a/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go b/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go index 70e6b5b823..15a5cfaf05 100644 --- a/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go +++ b/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go @@ -318,6 +318,32 @@ func requireSharedWrite(ctx context.Context, t *testing.T, router *e2e.RouterCli requireContentAtBoth(ctx, t, router, actorRef, writePath, aliasPath, probeWrittenContent) } +// requireAssignedNode fails unless the actor's status records want as its +// assigned_node. An empty want means no node is recorded. +func requireAssignedNode(t *testing.T, actor *ateapipb.Actor, want string) { + t.Helper() + + if got := actor.GetStatus().GetAssignedNode(); got != want { + t.Errorf("actor assigned_node = %q, want %q", got, want) + } +} + +// requireAssignedToWorkerNode fails unless the running actor's assigned_node +// matches the node of its worker assignment. +func requireAssignedToWorkerNode(ctx context.Context, t *testing.T, clients *e2e.Clients, actorRef resources.ActorRef) { + t.Helper() + + actor, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{Actor: actorRef.ToObjectRef()}) + if err != nil { + t.Fatalf("GetActor: %v", err) + } + node := actor.GetStatus().GetWorkerAssignment().GetNodeName() + if node == "" { + t.Fatalf("running actor's worker assignment names no node: %v", actor.GetStatus().GetWorkerAssignment()) + } + requireAssignedNode(t, actor, node) +} + func TestCombinedVolumes(t *testing.T) { repo := os.Getenv("KO_DOCKER_REPO") if repo == "" { @@ -351,6 +377,9 @@ func TestCombinedVolumes(t *testing.T) { if _, err := e2e.ResumeActorAwaitCapacity(t, ctx, clients, &ateapipb.ResumeActorRequest{Actor: actorRef.ToObjectRef()}); err != nil { t.Fatalf("ResumeActor: %v", err) } + if storageClass != "" { + requireAssignedToWorkerNode(ctx, t, clients, actorRef) + } router, err := e2e.NewRouterClient(ctx) if err != nil { @@ -392,9 +421,13 @@ func TestCombinedVolumes(t *testing.T) { }) t.Run("SurvivesSuspendResume", func(t *testing.T) { - if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: actorRef.ToObjectRef()}); err != nil { + suspended, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: actorRef.ToObjectRef()}) + if err != nil { t.Fatalf("SuspendActor: %v", err) } + if storageClass != "" { + requireAssignedNode(t, suspended.GetActor(), "") + } // This suite checks volume restore, which can outlast the router's // request-parking budget on a busy node. Resume through the control @@ -414,6 +447,7 @@ func TestCombinedVolumes(t *testing.T) { t.Skipf("StorageClass %q is not installed", e2e.StorageClass) } requireContentAtBoth(resumeCtx, t, router, actorRef, extPathA+"/multi.txt", extPathB+"/multi.txt", probeWrittenContent) + requireAssignedToWorkerNode(ctx, t, clients, actorRef) }) }) @@ -433,6 +467,9 @@ func TestCombinedVolumes(t *testing.T) { if got := reverted.GetActor().GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { t.Fatalf("state after revert = %v, want SUSPENDED", got) } + if storageClass != "" { + requireAssignedNode(t, reverted.GetActor(), "") + } resumeCtx, cancel := context.WithTimeout(ctx, 2*time.Minute) defer cancel()