diff --git a/cmd/ateapi/internal/controlapi/hosting_test.go b/cmd/ateapi/internal/controlapi/hosting_test.go index 585913fc95..8e6e0b130e 100644 --- a/cmd/ateapi/internal/controlapi/hosting_test.go +++ b/cmd/ateapi/internal/controlapi/hosting_test.go @@ -39,7 +39,10 @@ import ( // node must not count as hosting the actor in that window, nor once the worker // hosts another actor. func TestHostingNodeAfterWorkerRelease(t *testing.T) { - const node = "node1" + // The worker is the one wireTestAssignment names, so a pause or suspend + // reaches newWireCaptureWorkflow's fake atelet on its node. + wa := wireTestAssignment() + node := wa.GetNodeName() gvisorTemplate := &ateapipb.ActorTemplate{SandboxConfig: &ateapipb.SandboxConfig{SandboxClass: "gvisor"}} tests := []struct { name string @@ -49,9 +52,9 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { // workerClass is the worker's sandbox class. A template asks for // gVisor, so "microvm" makes the worker ineligible. workerClass string - // finalize runs the step that releases the worker and then records - // the actor's new state against st. - finalize func(ctx context.Context, st store.Interface, actorRef resources.ActorRef, actor *ateapipb.Actor) error + // finalize runs the workflow or step that releases the worker and + // then records the actor's new state against st. + finalize func(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, actor *ateapipb.Actor) error // wantRetry is the code of the retried finalize against the real // store. wantRetry codes.Code @@ -68,8 +71,11 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { return newLocalSnapshot(1, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "local-snap-1", "") }, workerClass: "gvisor", - finalize: func(ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error { - _, err := (&ActorWorkflow{store: st}).ensurePausedFinalized(ctx, actorRef) + // The actor is seeded SUSPENDING, so the pause skips its mark step + // and its first actor update is the finalize commit. + finalize: func(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error { + w, _ := newWireCaptureWorkflow(t, st) + _, err := w.PauseActor(ctx, actorRef) return err }, wantRetry: codes.OK, @@ -84,8 +90,10 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { return newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snapshot-1", someActorSnapshotURI(t, testStorageLocation, "team-a", "snapshot-1"), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) }, workerClass: "gvisor", - finalize: func(ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error { - _, err := (&ActorWorkflow{store: st}).ensureSuspendedFinalized(ctx, actorRef) + // As for the pause, the finalize commit is the first actor update. + finalize: func(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error { + w, _ := newWireCaptureWorkflow(t, st) + _, err := w.SuspendActor(ctx, actorRef) return err }, wantRetry: codes.OK, @@ -95,7 +103,7 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { name: "crash", state: ateapipb.ActorState_ACTOR_STATE_RUNNING, workerClass: "gvisor", - finalize: func(ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error { + finalize: func(_ *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error { return crashActor(ctx, st, actorRef, ateattr.OperationResume, "test crash") }, wantRetry: codes.OK, @@ -108,7 +116,7 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { name: "resume onto an ineligible worker", state: ateapipb.ActorState_ACTOR_STATE_RESUMING, workerClass: "microvm", - finalize: func(ctx context.Context, st store.Interface, actorRef resources.ActorRef, actor *ateapipb.Actor) error { + finalize: func(_ *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, actor *ateapipb.Actor) error { _, err := (&ActorWorkflow{store: st, scheduler: scheduling.New(nil)}).validateAssignedWorker(ctx, actorRef, actor, gvisorTemplate) return err }, @@ -120,14 +128,15 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { t.Run(tc.name, func(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} - workerName := testWorkerUID("worker-pod-1") + workerName := wa.GetWorker().GetName() if _, err := persistence.CreateWorker(ctx, &ateapipb.Worker{ Metadata: &ateapipb.ResourceMetadata{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, + WorkerNamespace: wa.GetWorkerNamespace(), + WorkerPool: wa.GetWorkerPool(), + WorkerPod: wa.GetWorkerPod(), + WorkerPodUid: wa.GetWorkerPodUid(), NodeName: node, SandboxClass: tc.workerClass, Status: &ateapipb.WorkerStatus{State: ateapipb.WorkerState_WORKER_STATE_ACTIVE, Capacity: &ateapipb.WorkerResources{Actors: 1}}, @@ -135,24 +144,18 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { t.Fatalf("CreateWorker: %v", err) } status := &ateapipb.ActorStatus{ - State: tc.state, - AssignedNode: node, - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - NodeName: node, - }, + State: tc.state, + AssignedNode: node, + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, } if tc.snapshot != nil { status.Snapshots = []*ateapipb.Snapshot{tc.snapshot(t)} } created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - Status: status, + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), + Status: status, }) uid := created.GetMetadata().GetUid() seedAssignment(t, persistence, workerName, &ateapipb.ActorAssignment{ @@ -181,7 +184,7 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { } // The release commits; recording the new state fails. - if err := tc.finalize(ctx, conflictingUpdateStore{Interface: persistence}, actorRef, actor); err == nil { + if err := tc.finalize(t, ctx, conflictingUpdateStore{Interface: persistence}, actorRef, actor); err == nil { t.Fatal("finalize succeeded through a store that fails every actor update") } if _, err := persistence.GetWorkerAssignment(ctx, workerName, uid); !errors.Is(err, store.ErrNotFound) { @@ -205,7 +208,7 @@ func TestHostingNodeAfterWorkerRelease(t *testing.T) { } // A retry records the new state. - if err := tc.finalize(ctx, persistence, actorRef, actor); apierror.Code(err) != tc.wantRetry { + if err := tc.finalize(t, ctx, persistence, actorRef, actor); apierror.Code(err) != tc.wantRetry { t.Fatalf("finalize retry: code %v, want %v (err: %v)", apierror.Code(err), tc.wantRetry, err) } got, actor = hostingNode("finalized") diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index 048b60ebb3..7a90002b3f 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -28,6 +28,7 @@ import ( "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func TestDeleteActorWorkflow_ExecutionPaths(t *testing.T) { @@ -444,40 +445,21 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) { objects := objectstoretest.New() actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} - workerName := testWorkerUID("pod-1") + workerName := wireTestAssignment().GetWorker().GetName() actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, ActorTemplate: &ateapipb.ObjectRef{Atespace: "team-a", Name: "sub-tmpl"}, Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_RUNNING, - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "worker-ns", - WorkerPool: "pool", - WorkerPod: "pod-1", - WorkerPodUid: workerName, - }, + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + WorkerAssignment: wireTestAssignment(), }, }) - if _, err := persistence.CreateWorker(ctx, &ateapipb.Worker{ - Metadata: &ateapipb.ResourceMetadata{Name: workerName}, - WorkerNamespace: "worker-ns", - WorkerPool: "pool", - WorkerPod: "pod-1", - WorkerPodUid: workerName, - Status: &ateapipb.WorkerStatus{}, - }); err != nil { - t.Fatalf("CreateWorker: %v", err) - } - seedAssignment(t, persistence, workerName, &ateapipb.ActorAssignment{ - Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorUid: actor.GetMetadata().GetUid(), - }) + seedWireWorker(t, ctx, persistence, assignmentOf(actor)) if tt.hasBeenSuspended { previous := mustActorSnapshotURI(t, template, actor, "old") objects.PutSnapshot(t, previous, "manifest.json") - actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { + mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { s.LastAssignedGeneration = 1 s.Snapshots = []*ateapipb.Snapshot{ newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", previous.Name(), previous.String(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED), @@ -485,21 +467,30 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) { }) } - actorWorkflow := NewActorWorkflow(persistence, nil, nil, nil, nil, nil, "", nil, objectstoreplugintest.ControlClient(objects)) - // Suspend the actor as far as it gets: MarkSuspending mints the - // in-progress URI, and the checkpoint writes under it - actor, err := actorWorkflow.ensureMarkedSuspending(ctx, actorRef, actor, template) - if err != nil { - t.Fatalf("ensureMarkedSuspending: %v", err) + actorWorkflow, atelet := newWireCaptureWorkflow(t, persistence) + actorWorkflow.objectStorePlugin = objectstoreplugintest.ControlClient(objects) + // atelet writes part of the snapshot, then the worker's pod goes + // away mid-checkpoint, so the checkpoint fails and the suspend never + // commits. + var fresh resources.SnapshotURI + atelet.onCheckpoint = func(context.Context) error { + uri, err := resources.ParseSnapshotURI(atelet.checkpointRequest().GetSnapshot().GetObject().GetSnapshotUri()) + if err != nil { + t.Errorf("ParseSnapshotURI: %v", err) + return err + } + fresh = uri + objects.PutSnapshot(t, fresh, "manifest.json") + // The worker's pod goes away with the commit still outstanding, so + // the suspend never gets to finish. + if _, err := NewWorkerWorkflow(persistence).DeleteWorker(ctx, workerName, store.DeletePreconditions{}); err != nil { + t.Errorf("DeleteWorker: %v", err) + return err + } + return status.Error(codes.Unavailable, "worker pod is gone") } - freshSnap := findLatestDurableSnapshot(actor.GetStatus(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) - fresh := mustParseSnapshotURI(t, freshSnap.GetDurableSnapshot().GetObject().GetSnapshotUri()) - objects.PutSnapshot(t, fresh, "manifest.json") - - // The worker's pod goes away with the commit still outstanding, so - // the suspend never gets to finish. - if _, err := NewWorkerWorkflow(persistence).DeleteWorker(ctx, workerName, store.DeletePreconditions{}); err != nil { - t.Fatalf("DeleteWorker: %v", err) + if _, err := actorWorkflow.SuspendActor(ctx, actorRef); apierror.Code(err) != codes.Unavailable { + t.Fatalf("SuspendActor err = %v, want Unavailable from the failed checkpoint", err) } // The actor is CRASHED; DeleteActor deletes the in-progress snapshot diff --git a/cmd/ateapi/internal/controlapi/workflow_pause_test.go b/cmd/ateapi/internal/controlapi/workflow_pause_test.go index 2a44bb8235..76be4a5088 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_pause_test.go @@ -16,6 +16,7 @@ package controlapi import ( "context" + "fmt" "testing" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" @@ -29,50 +30,44 @@ import ( "google.golang.org/protobuf/testing/protocmp" ) -// TestEnsurePausedFinalized_WorkerGone reproduces the scenario where the worker -// pod disappears from the DB during pause finalization, so the node it ran on -// is unknown. +// TestPauseActor_Finalized_WorkerGone reproduces the scenario where the +// worker pod disappears from the DB during pause finalization, so the node it +// ran on is unknown. // // Current behavior: AssignedNode is left empty, and the actor is crashed // instead of left paused, since a local snapshot with an unknown node can -// never be safely resumed. The step fails with FailedPrecondition rather than +// never be safely resumed. The pause fails with FailedPrecondition rather than // reporting the pause as done. -func TestEnsurePausedFinalized_WorkerGone(t *testing.T) { - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() - +func TestPauseActor_Finalized_WorkerGone(t *testing.T) { ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, nil) + w, atelet := newWireCaptureWorkflow(t, st) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} records := crashRecords(t) - actor := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + storetest.MustCreateActor(t, ctx, st, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, - WorkerAssignment: &ateapipb.WorkerAssignment{ - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - }, + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{ newLocalSnapshot(1, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "local-snap-1", ""), }, }, - } - storetest.MustCreateActor(t, ctx, st, actor) + }) // Intentionally NOT creating the worker in store, simulates worker already gone. - w := &ActorWorkflow{store: st} - if _, err := w.ensurePausedFinalized(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { - t.Fatalf("ensurePausedFinalized error = %v, want FailedPrecondition for the actor it crashed", err) + if _, err := w.PauseActor(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("PauseActor error = %v, want FailedPrecondition for the actor it crashed", err) } - - got, err := st.GetActor(ctx, actorRef) - if err != nil { - t.Fatalf("GetActor: %v", err) + if checkpoints, _ := atelet.calls(); checkpoints != 1 { + t.Errorf("Checkpoint calls = %d, want 1 before finalize", checkpoints) } + got := mustReadActor(t, ctx, st, actorRef) if got.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_CRASHED { t.Errorf("state = %v, want CRASHED (node name unknown, cannot resume safely)", got.GetStatus().GetState()) } @@ -99,45 +94,40 @@ func TestEnsurePausedFinalized_WorkerGone(t *testing.T) { } } -// TestEnsurePausedFinalized_AlreadyCrashed verifies that if the actor was -// already crashed out-of-band when ensurePausedFinalized runs with no -// AssignedNode, its existing Crash status and snapshots are preserved, no -// duplicate crash log record is emitted, and the step still fails. -func TestEnsurePausedFinalized_AlreadyCrashed(t *testing.T) { - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() - +// TestPauseActor_Finalized_AlreadyCrashed verifies that if the actor was +// crashed out-of-band while it checkpointed, and has no AssignedNode, the +// pause's finalize preserves the existing Crash status and snapshots, emits no +// duplicate crash log record, and still fails. +func TestPauseActor_Finalized_AlreadyCrashed(t *testing.T) { ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, nil) + w, atelet := newWireCaptureWorkflow(t, st) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} records := crashRecords(t) - originalCrash := newActorCrash(ateattr.OperationPause, "original crash reason") - actor := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + storetest.MustCreateActor(t, ctx, st, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_CRASHED, - Crash: originalCrash, - WorkerAssignment: &ateapipb.WorkerAssignment{ - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - }, + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{ newLocalSnapshot(1, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "local-snap-1", ""), }, }, - } - storetest.MustCreateActor(t, ctx, st, actor) + }) + originalCrash := newActorCrash(ateattr.OperationPause, "original crash reason") + updateDuringCheckpoint(t, atelet, st, actorRef, func(s *ateapipb.ActorStatus) { + s.State = ateapipb.ActorState_ACTOR_STATE_CRASHED + s.Crash = originalCrash + }) - w := &ActorWorkflow{store: st} - if _, err := w.ensurePausedFinalized(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { - t.Fatalf("ensurePausedFinalized error = %v, want FailedPrecondition for a CRASHED actor", err) - } - got, err := st.GetActor(ctx, actorRef) - if err != nil { - t.Fatalf("GetActor: %v", err) + if _, err := w.PauseActor(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("PauseActor error = %v, want FailedPrecondition for a CRASHED actor", err) } + got := mustReadActor(t, ctx, st, actorRef) if got.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_CRASHED { t.Errorf("state = %v, want CRASHED", got.GetStatus().GetState()) } @@ -155,59 +145,36 @@ func TestEnsurePausedFinalized_AlreadyCrashed(t *testing.T) { } } -// TestEnsurePausedFinalized_KeepsCrashFromWorkerDelete verifies that a crash -// the DeleteWorker workflow commits while ensurePausedFinalized is freeing the -// worker is kept: the step's update is pinned to the actor version it checked, -// so it fails Aborted instead of committing SUSPENDED over the crash. -func TestEnsurePausedFinalized_KeepsCrashFromWorkerDelete(t *testing.T) { - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() +// TestPauseActor_Finalized_KeepsCrashFromWorkerDelete verifies that a crash the +// DeleteWorker workflow commits while the pause's finalize is freeing the +// worker is kept: the finalize update is pinned to the actor version it +// checked, so it fails Aborted instead of committing SUSPENDED over the crash. +func TestPauseActor_Finalized_KeepsCrashFromWorkerDelete(t *testing.T) { ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, nil) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} - workerName := testWorkerUID("worker-pod-1") created := storetest.MustCreateActor(t, ctx, st, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_RUNNING, - AssignedNode: "node1", - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - }, + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), }, }) - if _, err := st.CreateWorker(ctx, &ateapipb.Worker{ - Metadata: &ateapipb.ResourceMetadata{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - NodeName: "node1", - Status: &ateapipb.WorkerStatus{}, - }); err != nil { - t.Fatalf("CreateWorker: %v", err) - } - seedAssignment(t, st, workerName, &ateapipb.ActorAssignment{ - Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorUid: created.GetMetadata().GetUid(), - }) - if _, err := (&ActorWorkflow{store: st}).ensureMarkedPausing(ctx, actorRef, created, &ateapipb.ActorTemplate{}); err != nil { - t.Fatalf("ensureMarkedPausing: %v", err) - } + seedWireWorker(t, ctx, st, assignmentOf(created)) - w := &ActorWorkflow{store: workerDeletedDuringReleaseStore{Interface: st}} - _, err := w.ensurePausedFinalized(ctx, actorRef) + w, _ := newWireCaptureWorkflow(t, workerDeletedDuringReleaseStore{Interface: st}) + _, err := w.PauseActor(ctx, actorRef) assertConcurrentCrashKept(t, ctx, st, actorRef, err, "pause failed: "+crashMessageWorkerPodGone) } -// TestEnsurePausedFinalized_RecordsFidelity verifies pause records the -// template's preferred_fidelity on the local Snapshot alongside Locality and -// leaves DurableSnapshot unset until the actor is suspended. -func TestEnsurePausedFinalized_RecordsFidelity(t *testing.T) { +// TestPauseActor_Finalized_RecordsFidelity verifies pause records the template's +// preferred_fidelity on the local Snapshot alongside Locality and leaves +// DurableSnapshot unset until the actor is suspended. +func TestPauseActor_Finalized_RecordsFidelity(t *testing.T) { tests := []struct { name string preferred ateapipb.SnapshotFidelity @@ -216,17 +183,18 @@ func TestEnsurePausedFinalized_RecordsFidelity(t *testing.T) { {"VOLUMES", ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES}, {"ROOTFS", ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_ROOTFS, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_ROOTFS}, {"MEMORY", ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}, - {"UNSPECIFIED recorded as UNSPECIFIED", ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_UNSPECIFIED, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_UNSPECIFIED}, + // The store defaults an unset preferred_fidelity to MEMORY. + {"UNSPECIFIED defaults to MEMORY", ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_UNSPECIFIED, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, &ateapipb.SnapshotConfig{PreferredFidelity: tc.preferred}) + w, _ := newWireCaptureWorkflow(t, st) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} - workerName := testWorkerUID("worker-pod-1") wantRuntime := &ateapipb.SandboxRuntime{ SandboxClass: "gvisor", Version: &ateapipb.VersionedSandboxCompat{ @@ -236,49 +204,22 @@ func TestEnsurePausedFinalized_RecordsFidelity(t *testing.T) { }, }, } + assignment := wireTestAssignment() + assignment.SandboxRuntime = wantRuntime created := storetest.MustCreateActor(t, ctx, st, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_RUNNING, - AssignedNode: "node1", - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - SandboxRuntime: wantRuntime, - }, + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + AssignedNode: "node-1", + WorkerAssignment: assignment, }, }) - if _, err := st.CreateWorker(ctx, &ateapipb.Worker{ - Metadata: &ateapipb.ResourceMetadata{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - NodeName: "node1", - SandboxClass: "gvisor", - }); err != nil { - t.Fatalf("CreateWorker: %v", err) - } - seedAssignment(t, st, workerName, &ateapipb.ActorAssignment{ - Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorUid: created.GetMetadata().GetUid(), - }) + seedWireWorker(t, ctx, st, assignmentOf(created)) - w := &ActorWorkflow{store: st} - tmpl := &ateapipb.ActorTemplate{ - SnapshotConfig: &ateapipb.SnapshotConfig{ - PreferredFidelity: tc.preferred, - }, - } - if _, err := w.ensureMarkedPausing(ctx, actorRef, created, tmpl); err != nil { - t.Fatalf("ensureMarkedPausing: %v", err) - } - got, err := w.ensurePausedFinalized(ctx, actorRef) + got, err := w.PauseActor(ctx, actorRef) if err != nil { - t.Fatalf("ensurePausedFinalized: %v", err) + t.Fatalf("PauseActor: %v", err) } if got.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { @@ -294,71 +235,59 @@ func TestEnsurePausedFinalized_RecordsFidelity(t *testing.T) { if localSnap.GetDurableSnapshot() != nil { t.Errorf("local DurableSnapshot = %v, want nil", localSnap.GetDurableSnapshot()) } - if localSnap.GetLocality() != "node1" { - t.Errorf("local Snapshot.Locality = %q, want %q", localSnap.GetLocality(), "node1") + if localSnap.GetLocality() != "node-1" { + t.Errorf("local Snapshot.Locality = %q, want %q", localSnap.GetLocality(), "node-1") } if diff := cmp.Diff(wantRuntime, localSnap.GetSandboxRuntime(), protocmp.Transform()); diff != "" { t.Errorf("Snapshot.SandboxRuntime mismatch (-want +got):\n%s", diff) } - if got.GetStatus().GetAssignedNode() != "node1" { - t.Errorf("AssignedNode = %q, want %q", got.GetStatus().GetAssignedNode(), "node1") + if got.GetStatus().GetAssignedNode() != "node-1" { + t.Errorf("AssignedNode = %q, want %q", got.GetStatus().GetAssignedNode(), "node-1") + } + if got.GetStatus().GetWorkerAssignment() != nil { + t.Errorf("WorkerAssignment = %v, want nil", got.GetStatus().GetWorkerAssignment()) + } + if a := firstAssignment(t, st, wireTestAssignment().GetWorker().GetName()); a != nil { + t.Errorf("worker assignment = %v, want released", a) } }) } } -// TestEnsurePausedFinalized_MissingLocalSnapshotClearsAssignment verifies that -// if the latest generation snapshot entry is missing when ensurePausedFinalized -// runs, the actor transitions to CRASHED and clears WorkerAssignment after -// releasing the worker rather than getting stuck SUSPENDING, and the step -// fails. -func TestEnsurePausedFinalized_MissingLocalSnapshotClearsAssignment(t *testing.T) { - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() +// TestPauseActor_Finalized_MissingLocalSnapshotClearsAssignment verifies that if the +// latest generation snapshot entry is missing when the pause finalizes, the +// actor transitions to CRASHED and clears WorkerAssignment after releasing +// the worker rather than getting stuck SUSPENDING, and the pause fails. +func TestPauseActor_Finalized_MissingLocalSnapshotClearsAssignment(t *testing.T) { ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, nil) + w, atelet := newWireCaptureWorkflow(t, st) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} records := crashRecords(t) - workerName := testWorkerUID("worker-pod-1") created := storetest.MustCreateActor(t, ctx, st, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, - AssignedNode: "node1", - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - }, + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, + Snapshots: []*ateapipb.Snapshot{ + newLocalSnapshot(1, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "local-snap-1", ""), + }, }, }) - if _, err := st.CreateWorker(ctx, &ateapipb.Worker{ - Metadata: &ateapipb.ResourceMetadata{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - NodeName: "node1", - Status: &ateapipb.WorkerStatus{}, - }); err != nil { - t.Fatalf("CreateWorker: %v", err) - } - seedAssignment(t, st, workerName, &ateapipb.ActorAssignment{ - Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorUid: created.GetMetadata().GetUid(), + seedWireWorker(t, ctx, st, assignmentOf(created)) + updateDuringCheckpoint(t, atelet, st, actorRef, func(s *ateapipb.ActorStatus) { + s.Snapshots = nil }) - w := &ActorWorkflow{store: st} - if _, err := w.ensurePausedFinalized(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { - t.Fatalf("ensurePausedFinalized error = %v, want FailedPrecondition for the actor it crashed", err) - } - got, err := st.GetActor(ctx, actorRef) - if err != nil { - t.Fatalf("GetActor: %v", err) + if _, err := w.PauseActor(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("PauseActor error = %v, want FailedPrecondition for the actor it crashed", err) } + got := mustReadActor(t, ctx, st, actorRef) if got.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_CRASHED { t.Errorf("state = %v, want CRASHED", got.GetStatus().GetState()) } @@ -388,8 +317,8 @@ func TestPauseActorWorkflow_RejectedAndIdempotentPaths(t *testing.T) { wantState ateapipb.ActorState }{ { - // Pausing a SUSPENDED actor is rejected by MarkPausingStep's - // CheckPrerequisite and the actor's state is left untouched. + // Pausing a SUSPENDED actor is rejected by the pause's state gate + // and the actor's state is left untouched. name: "not running rejected", seedState: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, wantErr: true, @@ -438,84 +367,98 @@ func TestPauseActorWorkflow_RejectedAndIdempotentPaths(t *testing.T) { } } -// TestEnsureMarkedPausing_StateMatrix verifies the pause edge's state gating -// against every actor state: RUNNING takes the edge, everything else is -// rejected with FailedPrecondition. SUSPENDED is rejected whatever its -// snapshots: the orchestrator early-returns before this step for a paused -// actor. SUSPENDING without a pending node-local snapshot is a suspend in -// progress, which owns the actor; a pause in progress is covered by -// TestEnsureMarkedPausing_ReentrySkipsPauseInProgress. -func TestEnsureMarkedPausing_StateMatrix(t *testing.T) { - allowed := map[ateapipb.ActorState]bool{ - ateapipb.ActorState_ACTOR_STATE_RUNNING: true, - } +// TestPauseActor_MarkedPausing_StateMatrix verifies the pause edge's state gating against +// every actor state: RUNNING takes the edge, everything else is rejected with +// FailedPrecondition and left untouched. SUSPENDED is rejected whatever its +// snapshots short of the paused shape, which fast-forwards (see +// TestPauseActorWorkflow_RejectedAndIdempotentPaths). SUSPENDING without a +// pending node-local snapshot is a suspend in progress, which owns the actor; +// a pause in progress is covered by TestPauseActor_MarkedPausing_ReentrySkipsPauseInProgress. +func TestPauseActor_MarkedPausing_StateMatrix(t *testing.T) { + ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, nil) + w := newMarkOnlyWorkflow(st) for _, seedState := range allActorStates { - ctx := context.Background() - persistence := newTestPersistence(t) - w := &ActorWorkflow{store: persistence} - - actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} - - actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - Status: &ateapipb.ActorStatus{State: seedState}, + actorRef := resources.ActorRef{Atespace: "team-a", Name: fmt.Sprintf("id%d", seedState)} + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", seedState, func(a *ateapipb.Actor) { + a.Status.WorkerAssignment = wireTestAssignment() }) - marked, err := w.ensureMarkedPausing(ctx, actorRef, actor, &ateapipb.ActorTemplate{}) - assertPrerequisiteResult(t, seedState, err, allowed[seedState]) - if err == nil && marked.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { - t.Errorf("state %v: ensureMarkedPausing returned actor in %v, want SUSPENDING", seedState, marked.GetStatus().GetState()) + _, err := w.PauseActor(ctx, actorRef) + stored := mustReadActor(t, ctx, st, actorRef) + if seedState == ateapipb.ActorState_ACTOR_STATE_RUNNING { + assertStoppedAtDial(t, err) + if got := stored.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { + t.Errorf("state %v: stored state = %v, want SUSPENDING", seedState, got) + } + if !latestSnapshotIsPendingLocal(stored.GetStatus()) { + t.Errorf("state %v: snapshots = %v, want a pending local snapshot at the latest generation", seedState, stored.GetStatus().GetSnapshots()) + } + continue + } + if got := apierror.Code(err); got != codes.FailedPrecondition { + t.Errorf("state %v: code = %v, want FailedPrecondition (err: %v)", seedState, got, err) + } + if got := stored.GetStatus().GetState(); got != seedState { + t.Errorf("state %v: stored state = %v, want it untouched", seedState, got) } } } -// TestEnsureMarkedPausing_ReentrySkipsPauseInProgress verifies a re-entered -// pause recognizes its own mark, a SUSPENDING actor whose latest generation -// is a pending node-local snapshot, and skips without minting another -// snapshot, while a running-origin suspend in progress (pending durable -// snapshot) is rejected. -func TestEnsureMarkedPausing_ReentrySkipsPauseInProgress(t *testing.T) { +// TestPauseActor_MarkedPausing_ReentrySkipsPauseInProgress verifies a re-entered pause +// recognizes its own mark, a SUSPENDING actor whose latest generation is a +// pending node-local snapshot, and checkpoints under that snapshot without +// minting another, while a running-origin suspend in progress (pending +// durable snapshot) is rejected before atelet is called. +func TestPauseActor_MarkedPausing_ReentrySkipsPauseInProgress(t *testing.T) { ctx := context.Background() - persistence := newTestPersistence(t) - w := &ActorWorkflow{store: persistence} + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, nil) + w, atelet := newWireCaptureWorkflow(t, st) actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} - pausing := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING}, - } - withPauseInProgress()(pausing) - created := storetest.MustCreateActor(t, ctx, persistence, pausing) - marked, err := w.ensureMarkedPausing(ctx, actorRef, created, &ateapipb.ActorTemplate{}) + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDING, withPauseInProgress(), func(a *ateapipb.Actor) { + a.Status.AssignedNode = "node-1" + a.Status.WorkerAssignment = wireTestAssignment() + }) + paused, err := w.PauseActor(ctx, actorRef) if err != nil { - t.Fatalf("ensureMarkedPausing on a pause in progress: %v", err) + t.Fatalf("PauseActor on a pause in progress: %v", err) + } + if got := atelet.checkpointRequest().GetSnapshot().GetSnapshotUuid(); got != "pause-snap" { + t.Errorf("CheckpointRequest.Snapshot.SnapshotUuid = %q, want the persisted %q", got, "pause-snap") } - if got := marked.GetStatus().GetLastAssignedGeneration(); got != 1 { + if got := paused.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { + t.Errorf("state = %v, want SUSPENDED", got) + } + if got := paused.GetStatus().GetLastAssignedGeneration(); got != 1 { t.Errorf("last_assigned_generation = %d, want 1 (no new snapshot minted on re-entry)", got) } - if got := len(marked.GetStatus().GetSnapshots()); got != 1 { + if got := len(paused.GetStatus().GetSnapshots()); got != 1 { t.Errorf("len(snapshots) = %d, want 1", got) } - suspending := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: "id2"}, - Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, - LastAssignedGeneration: 1, - Snapshots: []*ateapipb.Snapshot{ - newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snap", "gs://snapshots/s", ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), - }, - }, - } - created = storetest.MustCreateActor(t, ctx, persistence, suspending) - _, err = w.ensureMarkedPausing(ctx, resources.ActorRef{Atespace: actorRef.Atespace, Name: "id2"}, created, &ateapipb.ActorTemplate{}) + suspendingRef := resources.ActorRef{Atespace: actorRef.Atespace, Name: "id2"} + seedWorkflowActor(t, ctx, st, suspendingRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDING, func(a *ateapipb.Actor) { + a.Status.AssignedNode = "node-1" + a.Status.WorkerAssignment = wireTestAssignment() + a.Status.LastAssignedGeneration = 1 + a.Status.Snapshots = []*ateapipb.Snapshot{ + newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snap", "gs://snapshots/s", ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), + } + }) + _, err = w.PauseActor(ctx, suspendingRef) if got := apierror.Code(err); got != codes.FailedPrecondition { - t.Errorf("ensureMarkedPausing on a suspend in progress: code = %v, want FailedPrecondition (err: %v)", got, err) + t.Errorf("PauseActor on a suspend in progress: code = %v, want FailedPrecondition (err: %v)", got, err) + } + if checkpoints, _ := atelet.calls(); checkpoints != 1 { + t.Errorf("Checkpoint calls = %d, want 1 (none for the suspend in progress)", checkpoints) } } -func TestEnsureAteletPaused_DialFailureLeavesActorRetryable(t *testing.T) { +func TestPauseActor_AteletPaused_DialFailureLeavesActorRetryable(t *testing.T) { tests := []struct { name string prevSnapshot string @@ -534,6 +477,7 @@ func TestEnsureAteletPaused_DialFailureLeavesActorRetryable(t *testing.T) { t.Run(tt.name, func(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, nil) var snapshots []*ateapipb.Snapshot if tt.prevSnapshot != "" { @@ -542,7 +486,8 @@ func TestEnsureAteletPaused_DialFailureLeavesActorRetryable(t *testing.T) { snapshots = append(snapshots, newLocalSnapshot(2, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "actor-1-never-written", "")) actor := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, WorkerAssignment: &ateapipb.WorkerAssignment{ @@ -555,17 +500,13 @@ func TestEnsureAteletPaused_DialFailureLeavesActorRetryable(t *testing.T) { Snapshots: snapshots, }, } - created := storetest.MustCreateActor(t, ctx, persistence, actor) + storetest.MustCreateActor(t, ctx, persistence, actor) - w := &ActorWorkflow{store: persistence, dialer: newDanglingDialer()} - if _, err := w.ensureAteletPaused(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, created, &ateapipb.ActorTemplate{}); err == nil { - t.Fatal("ensureAteletPaused: want error when atelet is unreachable, got nil") - } + w := newMarkOnlyWorkflow(persistence) + _, err := w.PauseActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) + assertStoppedAtDial(t, err) - stored, err := persistence.GetActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) - if err != nil { - t.Fatalf("GetActor: %v", err) - } + stored := mustReadActor(t, ctx, persistence, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) if stored.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { t.Errorf("state = %v, want unchanged SUSPENDING", stored.GetStatus().GetState()) } @@ -581,9 +522,9 @@ func TestEnsureAteletPaused_DialFailureLeavesActorRetryable(t *testing.T) { } // TestPauseActor_CrashesWhenPausingActorMissingWorkerPod verifies that a -// pausing actor with no worker pod recorded is moved to CRASHED by -// ensureAteletPaused's corrupted-assignment check and the pause fails with -// FailedPrecondition. +// pausing actor with no worker pod recorded is moved to CRASHED by the +// corrupted-assignment check before atelet is called, and the pause fails +// with FailedPrecondition. func TestPauseActor_CrashesWhenPausingActorMissingWorkerPod(t *testing.T) { ctx := context.Background() st, cleanup := storetest.SetupTestStore(t) @@ -609,43 +550,54 @@ func TestPauseActor_CrashesWhenPausingActorMissingWorkerPod(t *testing.T) { } } -// TestEnsureMarkedPausing_GoldenAtespaceRejected verifies golden actors -// cannot be paused: by design they can only be suspended (committed). -func TestEnsureMarkedPausing_GoldenAtespaceRejected(t *testing.T) { - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() - w := newTestActorWorkflow(t, st, "ns", "tmpl1") - - _, err := w.ensureMarkedPausing(context.Background(), - resources.ActorRef{Atespace: resources.GoldenActorAtespace, Name: "golden-1"}, - &ateapipb.Actor{Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_RUNNING}}, - &ateapipb.ActorTemplate{}) +// TestPauseActor_MarkedPausing_GoldenAtespaceRejected verifies golden actors cannot be +// paused: by design they can only be suspended (committed). +func TestPauseActor_MarkedPausing_GoldenAtespaceRejected(t *testing.T) { + ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, nil) + w := newMarkOnlyWorkflow(st) + + actorRef := resources.ActorRef{Atespace: resources.GoldenActorAtespace, Name: "golden-1"} + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_RUNNING, func(a *ateapipb.Actor) { + a.Status.AssignedNode = "node-1" + a.Status.WorkerAssignment = wireTestAssignment() + }) + _, err := w.PauseActor(ctx, actorRef) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code = %v (err %v), want FailedPrecondition", got, err) } + if got := mustReadActor(t, ctx, st, actorRef).GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_RUNNING { + t.Errorf("stored state = %v, want RUNNING untouched", got) + } } -// TestEnsureAteletPaused_ForwardsPreferredFidelity verifies that local pause -// checkpoints forward the template's PreferredFidelity to atelet. -func TestEnsureAteletPaused_ForwardsPreferredFidelity(t *testing.T) { +// TestPauseActor_AteletPaused_ForwardsPreferredFidelity verifies that local pause +// checkpoints forward the template's PreferredFidelity to atelet, and label the +// pause's lifecycle-op metric with it. +func TestPauseActor_AteletPaused_ForwardsPreferredFidelity(t *testing.T) { tests := []struct { preferred ateapipb.SnapshotFidelity want ateletpb.SnapshotFidelity }{ {ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}, {ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES, ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES}, - {ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_UNSPECIFIED, ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_UNSPECIFIED}, + // The store defaults an unset preferred_fidelity to MEMORY. + {ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_UNSPECIFIED, ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}, } for _, tc := range tests { t.Run(tc.preferred.String(), func(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{PreferredFidelity: tc.preferred}) w, atelet := newWireCaptureWorkflow(t, persistence) + inst, reader := newTestInstruments(t) + w.instruments = inst actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} - created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorTemplate: &ateapipb.ObjectRef{Atespace: "ns", Name: "tmpl1"}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, AssignedNode: "node-1", @@ -656,19 +608,9 @@ func TestEnsureAteletPaused_ForwardsPreferredFidelity(t *testing.T) { }, }, }) - tmpl := &ateapipb.ActorTemplate{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "ns", Name: "tmpl1"}, - SnapshotConfig: &ateapipb.SnapshotConfig{ - PreferredFidelity: tc.preferred, - }, - } - wireFidelity, err := w.ensureAteletPaused(ctx, actorRef, created, tmpl) - if err != nil { - t.Fatalf("ensureAteletPaused: %v", err) - } - if want := ateattr.SnapshotFidelityValue(tc.want); wireFidelity != want { - t.Errorf("wireFidelity = %q, want %q", wireFidelity, want) + if _, err := w.PauseActor(ctx, actorRef); err != nil { + t.Fatalf("PauseActor: %v", err) } req := atelet.checkpointRequest() if req == nil { @@ -683,9 +625,14 @@ func TestEnsureAteletPaused_ForwardsPreferredFidelity(t *testing.T) { if got := req.GetSnapshot().GetGeneration(); got != 1 { t.Errorf("CheckpointRequest.Snapshot.Generation = %d, want 1, the in-progress snapshot's", got) } + if got := req.GetSnapshot().GetObject(); got != nil { + t.Errorf("CheckpointRequest.Snapshot.Object = %v, want nil for a node-local checkpoint", got) + } if got := req.GetUploadConfig().GetMode(); got != ateletpb.UploadMode_UPLOAD_MODE_DEFERRED { t.Errorf("CheckpointRequest.UploadConfig.Mode = %v, want UPLOAD_MODE_DEFERRED", got) } + // The fidelity sent to atelet labels the pause's lifecycle-op metric. + assertLifecycleFidelity(t, reader, ateattr.SnapshotFidelityValue(tc.want)) }) } } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 089afad786..c089f36a12 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -1053,11 +1053,18 @@ func TestLoadActorForResume_RunningActorShortCircuits(t *testing.T) { type capturingAtelet struct { ateletpb.UnimplementedAteletServer - mu sync.Mutex - restore *ateletpb.RestoreRequest - run *ateletpb.RunRequest - checkpoint *ateletpb.CheckpointRequest - promote *ateletpb.PromoteSnapshotRequest + mu sync.Mutex + restore *ateletpb.RestoreRequest + run *ateletpb.RunRequest + checkpoint *ateletpb.CheckpointRequest + promote *ateletpb.PromoteSnapshotRequest + checkpoints int + promotes int + + // onCheckpoint, when set, runs before Checkpoint returns, standing in for + // whatever else happens while the workload checkpoints. Its error fails + // the RPC. + onCheckpoint func(ctx context.Context) error } func (f *capturingAtelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (*ateletpb.RestoreResponse, error) { @@ -1076,8 +1083,15 @@ func (f *capturingAtelet) Run(ctx context.Context, req *ateletpb.RunRequest) (*a func (f *capturingAtelet) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRequest) (*ateletpb.CheckpointResponse, error) { f.mu.Lock() - defer f.mu.Unlock() f.checkpoint = proto.Clone(req).(*ateletpb.CheckpointRequest) + f.checkpoints++ + hook := f.onCheckpoint + f.mu.Unlock() + if hook != nil { + if err := hook(ctx); err != nil { + return nil, err + } + } return &ateletpb.CheckpointResponse{}, nil } @@ -1085,9 +1099,17 @@ func (f *capturingAtelet) PromoteSnapshot(ctx context.Context, req *ateletpb.Pro f.mu.Lock() defer f.mu.Unlock() f.promote = proto.Clone(req).(*ateletpb.PromoteSnapshotRequest) + f.promotes++ return &ateletpb.PromoteSnapshotResponse{}, nil } +// calls returns how many Checkpoint and PromoteSnapshot RPCs were received. +func (f *capturingAtelet) calls() (checkpoints, promotes int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.checkpoints, f.promotes +} + func (f *capturingAtelet) checkpointRequest() *ateletpb.CheckpointRequest { f.mu.Lock() defer f.mu.Unlock() diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go index c7fe7ca6bb..ecf052fb4f 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go @@ -189,20 +189,25 @@ func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { // allocate generation 4 rather than reusing 2 or 3. running := mustUpdateActorStatus(t, ctx, st, reverted, func(s *ateapipb.ActorStatus) { s.State = ateapipb.ActorState_ACTOR_STATE_RUNNING + s.AssignedNode = "node-1" + s.WorkerAssignment = wireTestAssignment() }) - tmpl := &ateapipb.ActorTemplate{ - SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots/"}, - } - suspending, err := w.ensureMarkedSuspending(ctx, actorRef, running, tmpl) + seedWireWorker(t, ctx, st, assignmentOf(running)) + wire, atelet := newWireCaptureWorkflow(t, st) + suspended, err := wire.SuspendActor(ctx, actorRef) if err != nil { - t.Fatalf("ensureMarkedSuspending after revert: %v", err) + t.Fatalf("SuspendActor after revert: %v", err) } - if got := suspending.GetStatus().GetLastAssignedGeneration(); got != 4 { + if got := suspended.GetStatus().GetLastAssignedGeneration(); got != 4 { t.Errorf("last_assigned_generation after post-revert suspend = %d, want 4", got) } - if got := snapshotAtLatestGeneration(suspending.GetStatus()).GetGeneration(); got != 4 { + newSnap := snapshotAtLatestGeneration(suspended.GetStatus()) + if got := newSnap.GetGeneration(); got != 4 { t.Errorf("new snapshot generation after post-revert suspend = %d, want 4", got) } + if got, want := atelet.checkpointRequest().GetSnapshot().GetSnapshotUuid(), newSnap.GetUuid(); got != want { + t.Errorf("CheckpointRequest.Snapshot.SnapshotUuid = %q, want the generation 4 snapshot %q", got, want) + } }) } } diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go b/cmd/ateapi/internal/controlapi/workflow_suspend_test.go index 3530979584..8bed842237 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend_test.go @@ -17,15 +17,15 @@ package controlapi import ( "context" "errors" + "fmt" "strings" "testing" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" "github.com/agent-substrate/substrate/internal/apierror" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/installdefaults" - "github.com/agent-substrate/substrate/internal/objectstore/objectstoretest" - "github.com/agent-substrate/substrate/internal/objectstoreplugin/objectstoreplugintest" "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" @@ -36,10 +36,10 @@ import ( "k8s.io/client-go/tools/cache" ) -func TestEnsureMarkedSuspending_SnapshotURI(t *testing.T) { +func TestSuspendActor_MarkedSuspending_SnapshotURI(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) - workerName := testWorkerUID("worker-pod-1") + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: "gs://bucket/root/"}) wantRuntime := &ateapipb.SandboxRuntime{ SandboxClass: "gvisor", Version: &ateapipb.VersionedSandboxCompat{ @@ -49,28 +49,21 @@ func TestEnsureMarkedSuspending_SnapshotURI(t *testing.T) { }, }, } - actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + assignment := wireTestAssignment() + assignment.SandboxRuntime = wantRuntime + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_RUNNING, - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - SandboxRuntime: wantRuntime, - }, + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + WorkerAssignment: assignment, }, }) - tmpl := &ateapipb.ActorTemplate{ - SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://bucket/root/"}, - } - w := &ActorWorkflow{store: persistence} - marked, err := w.ensureMarkedSuspending(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, actor, tmpl) - if err != nil { - t.Fatalf("ensureMarkedSuspending: %v", err) - } + w := newMarkOnlyWorkflow(persistence) + _, err := w.SuspendActor(ctx, actorRef) + assertStoppedAtDial(t, err) + marked := mustReadActor(t, ctx, persistence, actorRef) inProgressSnap := findLatestDurableSnapshot(marked.GetStatus(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) uri := mustParseSnapshotURI(t, inProgressSnap.GetDurableSnapshot().GetObject().GetSnapshotUri()) @@ -89,31 +82,44 @@ func TestEnsureMarkedSuspending_SnapshotURI(t *testing.T) { } } -// TestEnsureMarkedSuspending_ReentryKeepsPersistedSnapshotLocation verifies a +// TestSuspendActor_MarkedSuspending_ReentryKeepsPersistedSnapshotLocation verifies a // re-entered workflow does not mint a second snapshot location: the location -// persisted by the first attempt stays authoritative. -func TestEnsureMarkedSuspending_ReentryKeepsPersistedSnapshotLocation(t *testing.T) { +// persisted by the first attempt stays authoritative, both on the wire and in +// the snapshot the suspend commits. +func TestSuspendActor_MarkedSuspending_ReentryKeepsPersistedSnapshotLocation(t *testing.T) { const firstAttempt = "gs://bucket/root/atespaces/team-a/actors/actor-uid/snapshots/first-attempt" ctx := context.Background() persistence := newTestPersistence(t) - actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) + w, atelet := newWireCaptureWorkflow(t, persistence) + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{ newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "first-attempt", firstAttempt, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), }, }, }) - w := &ActorWorkflow{store: persistence} - marked, err := w.ensureMarkedSuspending(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, actor, &ateapipb.ActorTemplate{}) + suspended, err := w.SuspendActor(ctx, actorRef) if err != nil { - t.Fatalf("ensureMarkedSuspending: %v", err) + t.Fatalf("SuspendActor: %v", err) + } + if got := atelet.checkpointRequest().GetSnapshot().GetObject().GetSnapshotUri(); got != firstAttempt { + t.Errorf("CheckpointRequest.Snapshot.Object.SnapshotUri = %q, want the first attempt's location", got) } - if gotSnap := findLatestDurableSnapshot(marked.GetStatus(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING); gotSnap.GetDurableSnapshot().GetObject().GetSnapshotUri() != firstAttempt { - t.Errorf("in-progress durable snapshot URI = %q, want the first attempt's location", gotSnap.GetDurableSnapshot().GetObject().GetSnapshotUri()) + if got := suspended.GetStatus().GetLastAssignedGeneration(); got != 1 { + t.Errorf("last_assigned_generation = %d, want 1 (no new snapshot minted on re-entry)", got) + } + gotSnap := findLatestDurableSnapshot(suspended.GetStatus(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED) + if got := gotSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(); got != firstAttempt { + t.Errorf("committed durable snapshot URI = %q, want the first attempt's location", got) } } @@ -171,88 +177,96 @@ func TestSuspendActorWorkflow_RejectedAndIdempotentPaths(t *testing.T) { } } -// TestEnsureMarkedSuspending_StateMatrix verifies the suspend edge's state -// gating against every actor state: RUNNING takes the edge (checkpoint the -// workload), SUSPENDING skips (a previous attempt already marked the actor), -// everything else is rejected with FailedPrecondition. SUSPENDED without a -// node-local snapshot is rejected here because the orchestrator -// early-returns before this step for a fully suspended actor; SUSPENDED with -// one is the paused shape, covered by -// TestEnsureMarkedSuspending_PausedPromotesLocalSnapshot. -func TestEnsureMarkedSuspending_StateMatrix(t *testing.T) { - allowed := map[ateapipb.ActorState]bool{ - ateapipb.ActorState_ACTOR_STATE_RUNNING: true, - ateapipb.ActorState_ACTOR_STATE_SUSPENDING: true, // skipped, not re-marked - } +// TestSuspendActor_MarkedSuspending_StateMatrix verifies the suspend edge's state gating +// against every actor state: RUNNING takes the edge (checkpoint the +// workload), SUSPENDING skips the mark (a previous attempt already marked the +// actor) and goes on to atelet, everything else is rejected with +// FailedPrecondition and left untouched. SUSPENDED without a node-local +// snapshot fast-forwards (see TestSuspendActorWorkflow_RejectedAndIdempotentPaths); +// SUSPENDED with one is the paused shape, covered by +// TestSuspendActor_MarkedSuspending_PausedPromotesLocalSnapshot. +func TestSuspendActor_MarkedSuspending_StateMatrix(t *testing.T) { + ctx := context.Background() + persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}) + w := newMarkOnlyWorkflow(persistence) for _, seedState := range allActorStates { - ctx := context.Background() - persistence := newTestPersistence(t) - w := &ActorWorkflow{store: persistence} - - actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} - actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - Status: &ateapipb.ActorStatus{State: seedState}, + actorRef := resources.ActorRef{Atespace: "team-a", Name: fmt.Sprintf("id%d", seedState)} + seedWorkflowActor(t, ctx, persistence, actorRef, "ns", "tmpl1", seedState, func(a *ateapipb.Actor) { + a.Status.WorkerAssignment = wireTestAssignment() }) - tmpl := &ateapipb.ActorTemplate{ - SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}, - } - marked, err := w.ensureMarkedSuspending(ctx, actorRef, actor, tmpl) - assertPrerequisiteResult(t, seedState, err, allowed[seedState]) - if err == nil && marked.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { - t.Errorf("state %v: ensureMarkedSuspending returned actor in %v, want SUSPENDING", seedState, marked.GetStatus().GetState()) + _, err := w.SuspendActor(ctx, actorRef) + stored := mustReadActor(t, ctx, persistence, actorRef) + switch seedState { + case ateapipb.ActorState_ACTOR_STATE_RUNNING, ateapipb.ActorState_ACTOR_STATE_SUSPENDING: + assertStoppedAtDial(t, err) + if got := stored.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { + t.Errorf("state %v: stored state = %v, want SUSPENDING", seedState, got) + } + case ateapipb.ActorState_ACTOR_STATE_SUSPENDED: + if err != nil { + t.Errorf("state %v: SuspendActor = %v, want the fast-forward", seedState, err) + } + if got := stored.GetStatus().GetState(); got != seedState { + t.Errorf("state %v: stored state = %v, want it untouched", seedState, got) + } + default: + if got := apierror.Code(err); got != codes.FailedPrecondition { + t.Errorf("state %v: code = %v, want FailedPrecondition (err: %v)", seedState, got, err) + } + if got := stored.GetStatus().GetState(); got != seedState { + t.Errorf("state %v: stored state = %v, want it untouched", seedState, got) + } } } } -// TestEnsureMarkedSuspending_RejectsPauseInProgress verifies a suspend does -// not adopt a SUSPENDING actor whose latest generation is a pending -// node-local snapshot: that actor is mid-pause, and the pause owns it until -// it is paused. -func TestEnsureMarkedSuspending_RejectsPauseInProgress(t *testing.T) { +// TestSuspendActor_MarkedSuspending_RejectsPauseInProgress verifies a suspend does not adopt a +// SUSPENDING actor whose latest generation is a pending node-local snapshot: +// that actor is mid-pause, and the pause owns it until it is paused. +func TestSuspendActor_MarkedSuspending_RejectsPauseInProgress(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) - w := &ActorWorkflow{store: persistence} + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}) + w := newMarkOnlyWorkflow(persistence) actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} - pausing := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING}, - } - withPauseInProgress()(pausing) - created := storetest.MustCreateActor(t, ctx, persistence, pausing) + seedWorkflowActor(t, ctx, persistence, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDING, withPauseInProgress(), func(a *ateapipb.Actor) { + a.Status.WorkerAssignment = wireTestAssignment() + }) - tmpl := &ateapipb.ActorTemplate{SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}} - _, err := w.ensureMarkedSuspending(ctx, actorRef, created, tmpl) + _, err := w.SuspendActor(ctx, actorRef) if got := apierror.Code(err); got != codes.FailedPrecondition { - t.Errorf("ensureMarkedSuspending on a pause in progress: code = %v, want FailedPrecondition (err: %v)", got, err) - } - stored, err := persistence.GetActor(ctx, actorRef) - if err != nil { - t.Fatalf("GetActor: %v", err) + t.Errorf("SuspendActor on a pause in progress: code = %v, want FailedPrecondition (err: %v)", got, err) } + stored := mustReadActor(t, ctx, persistence, actorRef) if got := stored.GetStatus().GetLastAssignedGeneration(); got != 1 { t.Errorf("last_assigned_generation = %d, want 1 (the pause's snapshot untouched)", got) } + if !latestSnapshotIsPendingLocal(stored.GetStatus()) { + t.Errorf("snapshots = %v, want the pause's pending local snapshot untouched", stored.GetStatus().GetSnapshots()) + } } -// TestEnsureMarkedSuspending_PausedPromotesLocalSnapshot verifies the suspend -// edge from a paused actor (SUSPENDED whose latest snapshot is node-local): -// it takes the edge and promotes that snapshot in place to a pending durable -// one, keeping its generation, uuid, and sandbox runtime stamp, with the -// upload destination derived from the uuid. -func TestEnsureMarkedSuspending_PausedPromotesLocalSnapshot(t *testing.T) { +// TestSuspendActor_MarkedSuspending_PausedPromotesLocalSnapshot verifies the suspend edge from +// a paused actor (SUSPENDED whose latest snapshot is node-local): it takes the +// edge and promotes that snapshot in place to a pending durable one, keeping +// its generation, uuid, and sandbox runtime stamp, with the upload destination +// derived from the uuid. +func TestSuspendActor_MarkedSuspending_PausedPromotesLocalSnapshot(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) - w := &ActorWorkflow{store: persistence} + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}) + w := newMarkOnlyWorkflow(persistence) actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} localSnap := newLocalSnapshot(1, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snap-1", "node-1") localSnap.SandboxRuntime = testSandboxRuntime() actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, AssignedNode: "node-1", @@ -261,13 +275,10 @@ func TestEnsureMarkedSuspending_PausedPromotesLocalSnapshot(t *testing.T) { }, }) - tmpl := &ateapipb.ActorTemplate{ - SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}, - } - marked, err := w.ensureMarkedSuspending(ctx, actorRef, actor, tmpl) - if err != nil { - t.Fatalf("ensureMarkedSuspending: %v", err) - } + // The upload's dial fails, so the actor stays as the mark left it. + _, err := w.SuspendActor(ctx, actorRef) + assertStoppedAtDial(t, err) + marked := mustReadActor(t, ctx, persistence, actorRef) if got := marked.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { t.Errorf("state = %v, want SUSPENDING", got) } @@ -299,9 +310,9 @@ func TestEnsureMarkedSuspending_PausedPromotesLocalSnapshot(t *testing.T) { } // TestSuspendActor_CrashesWhenSuspendingActorMissingWorkerPod verifies that a -// SUSPENDING actor with no worker pod recorded is moved to CRASHED by -// CallAteletSuspendStep's prerequisite check and the suspend fails with -// FailedPrecondition. +// SUSPENDING actor with no worker pod recorded is moved to CRASHED by the +// corrupted-assignment check before atelet is called, and the suspend fails +// with FailedPrecondition. func TestSuspendActor_CrashesWhenSuspendingActorMissingWorkerPod(t *testing.T) { ctx := context.Background() st, cleanup := storetest.SetupTestStore(t) @@ -343,7 +354,7 @@ func newDanglingDialer() *AteletDialer { return NewAteletDialer(empty, installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "") } -func TestEnsureAteletSuspended_DialFailureLeavesActorRetryable(t *testing.T) { +func TestSuspendActor_AteletSuspended_DialFailureLeavesActorRetryable(t *testing.T) { neverWritten := someActorSnapshotURI(t, testStorageLocation, "team-a", "never-written") tests := []struct { @@ -364,6 +375,7 @@ func TestEnsureAteletSuspended_DialFailureLeavesActorRetryable(t *testing.T) { t.Run(tt.name, func(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) var snapshots []*ateapipb.Snapshot if tt.prevSnapshot != "" { snapshots = append(snapshots, newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "prev-snapshot", tt.prevSnapshot, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED)) @@ -371,7 +383,8 @@ func TestEnsureAteletSuspended_DialFailureLeavesActorRetryable(t *testing.T) { snapshots = append(snapshots, newDurableSnapshot(2, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "never-written", neverWritten, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING)) actor := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, WorkerAssignment: &ateapipb.WorkerAssignment{ @@ -384,12 +397,11 @@ func TestEnsureAteletSuspended_DialFailureLeavesActorRetryable(t *testing.T) { Snapshots: snapshots, }, } - created := storetest.MustCreateActor(t, ctx, persistence, actor) + storetest.MustCreateActor(t, ctx, persistence, actor) - w := &ActorWorkflow{store: persistence, dialer: newDanglingDialer()} - if _, err := w.ensureAteletSuspended(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, created, &ateapipb.ActorTemplate{}); err == nil { - t.Fatal("ensureAteletSuspended: want error when atelet is unreachable, got nil") - } + w := newMarkOnlyWorkflow(persistence) + _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) + assertStoppedAtDial(t, err) // A dial failure is transient from this workflow's point of view: it // must not crash the actor or touch its snapshot state. A worker that @@ -412,20 +424,23 @@ func TestEnsureAteletSuspended_DialFailureLeavesActorRetryable(t *testing.T) { } } -// TestEnsureAteletSuspended_WireRequest verifies a running-origin suspend -// checkpoints to the actor's in-progress durable snapshot with an immediate -// upload, since FinalizeSuspended then marks that snapshot COMPLETED. -func TestEnsureAteletSuspended_WireRequest(t *testing.T) { +// TestSuspendActor_AteletSuspended_WireRequest verifies a running-origin suspend checkpoints +// to the actor's in-progress durable snapshot with an immediate upload, since +// the suspend's finalize then marks that snapshot COMPLETED. +func TestSuspendActor_AteletSuspended_WireRequest(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{PreferredFidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}) w, atelet := newWireCaptureWorkflow(t, persistence) + inst, reader := newTestInstruments(t) + w.instruments = inst actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} snapshotName := "2026-01-01t00-00-00z-abc" snapshotURI := someActorSnapshotURI(t, testStorageLocation, "team-a", snapshotName) - created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorTemplate: &ateapipb.ObjectRef{Atespace: "ns", Name: "tmpl1"}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, AssignedNode: "node-1", @@ -436,13 +451,9 @@ func TestEnsureAteletSuspended_WireRequest(t *testing.T) { }, }, }) - tmpl := &ateapipb.ActorTemplate{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "ns", Name: "tmpl1"}, - SnapshotConfig: &ateapipb.SnapshotConfig{PreferredFidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}, - } - if _, err := w.ensureAteletSuspended(ctx, actorRef, created, tmpl); err != nil { - t.Fatalf("ensureAteletSuspended: %v", err) + if _, err := w.SuspendActor(ctx, actorRef); err != nil { + t.Fatalf("SuspendActor: %v", err) } req := atelet.checkpointRequest() if req == nil { @@ -461,38 +472,60 @@ func TestEnsureAteletSuspended_WireRequest(t *testing.T) { if got := req.GetUploadConfig(); !proto.Equal(got, wantConfig) { t.Errorf("CheckpointRequest.UploadConfig = %v, want %v", got, wantConfig) } + // The fidelity sent to atelet labels the suspend's lifecycle-op metric. + assertLifecycleFidelity(t, reader, ateattr.SnapshotFidelityMemory) } -// TestEnsureSuspendedFinalized_NoAssignment verifies finalization runs even when -// the actor has no worker assignment: the external snapshot must be recorded on -// the actor and the actor moved to SUSPENDED rather than silently left -// SUSPENDING. This is the shape a paused-origin suspend (#791) produces — a -// paused actor has no worker — and the regression test for finalization +// TestSuspendActor_Finalized_NoAssignment verifies finalization runs even +// when the actor has no worker assignment: the external snapshot must be +// recorded on the actor and the actor moved to SUSPENDED rather than silently +// left SUSPENDING. This is the shape a paused-origin suspend (#791) produces — +// a paused actor has no worker — and the regression test for finalization // previously living inside the worker-freeing branch. -func TestEnsureSuspendedFinalized_NoAssignment(t *testing.T) { +func TestSuspendActor_Finalized_NoAssignment(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) + w, atelet := newWireCaptureWorkflow(t, persistence) + inst, reader := newTestInstruments(t) + w.instruments = inst + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} snapshotURI := someActorSnapshotURI(t, testStorageLocation, "team-a", "2026-01-01t00-00-00z-abc") snap := newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "tmpl-uid-1", "actor-1-pause-snapshot", snapshotURI, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) - snap.Locality = "node1" - actor := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + snap.Locality = "node-1" + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, - AssignedNode: "node1", + AssignedNode: "node-1", LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{snap}, }, - } - storetest.MustCreateActor(t, ctx, persistence, actor) + }) - w := &ActorWorkflow{store: persistence} - stored, err := w.ensureSuspendedFinalized(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) + stored, err := w.SuspendActor(ctx, actorRef) if err != nil { - t.Fatalf("ensureSuspendedFinalized: %v", err) + t.Fatalf("SuspendActor: %v", err) } + // With no worker there is no workload to checkpoint: the suspend uploads + // the node-local snapshot instead. + if checkpoints, promotes := atelet.calls(); checkpoints != 0 || promotes != 1 { + t.Errorf("atelet calls = %d Checkpoint, %d PromoteSnapshot; want 0 and 1", checkpoints, promotes) + } + promote := atelet.promoteRequest() + if got := promote.GetSnapshot().GetSnapshotUuid(); got != "actor-1-pause-snapshot" { + t.Errorf("PromoteSnapshotRequest.Snapshot.SnapshotUuid = %q, want %q", got, "actor-1-pause-snapshot") + } + if got := promote.GetSnapshot().GetObject().GetSnapshotUri(); got != snapshotURI { + t.Errorf("PromoteSnapshotRequest.Snapshot.Object.SnapshotUri = %q, want %q", got, snapshotURI) + } + // The template's preferred fidelity, which the store defaults to MEMORY, + // labels the suspend's lifecycle-op metric. + assertLifecycleFidelity(t, reader, ateattr.SnapshotFidelityMemory) + if stored.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { t.Errorf("state = %v, want SUSPENDED", stored.GetStatus().GetState()) } @@ -516,11 +549,11 @@ func TestEnsureSuspendedFinalized_NoAssignment(t *testing.T) { } } -// TestEnsureSuspendedFinalized_ReleasesReplacedSnapshot verifies which external +// TestSuspendActor_Finalized_ReleasesReplacedSnapshot verifies which external // snapshot a suspend collects. The actor's previous one goes away, because // nothing else can name it — unless the actor was only borrowing it from a tag, // in which case the tag owns those objects and the suspend must leave them be. -func TestEnsureSuspendedFinalized_ReleasesReplacedSnapshot(t *testing.T) { +func TestSuspendActor_Finalized_ReleasesReplacedSnapshot(t *testing.T) { // The name of the snapshot the suspend has just written, in place of the // one the actor was running from. const snapshotName = "2026-01-01t00-00-00z-new" @@ -549,13 +582,17 @@ func TestEnsureSuspendedFinalized_ReleasesReplacedSnapshot(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") - w, objects := newFinalizeWorkflow(persistence) + w, _, objects := newWireFinalizeWorkflow(t, persistence) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, ActorTemplate: &ateapipb.ObjectRef{Atespace: "team-a", Name: "sub-tmpl"}, - Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING}, + Status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), + }, }) previous := mustActorSnapshotURI(t, template, actor, "old") @@ -576,9 +613,9 @@ func TestEnsureSuspendedFinalized_ReleasesReplacedSnapshot(t *testing.T) { } }) - stored, err := w.ensureSuspendedFinalized(ctx, actorRef) + stored, err := w.SuspendActor(ctx, actorRef) if err != nil { - t.Fatalf("ensureSuspendedFinalized: %v", err) + t.Fatalf("SuspendActor: %v", err) } // Whichever way the previous snapshot went, the actor now owns the // one it just wrote and is no longer borrowing. @@ -606,23 +643,27 @@ func isObjectStoreErr(err error) bool { return err != nil && strings.Contains(err.Error(), errObjectStore.Error()) } -// TestEnsureSuspendedFinalized_CommitsDespiteObjectStoreFailure verifies that +// TestSuspendActor_Finalized_CommitsDespiteObjectStoreFailure verifies that // failing to collect the snapshot a suspend replaced does not fail the // suspend. The worker is already released by then, so aborting would leave the // actor SUSPENDING with no way to resume; it commits SUSPENDED on the new // snapshot instead and leaves the replaced one in storage. -func TestEnsureSuspendedFinalized_CommitsDespiteObjectStoreFailure(t *testing.T) { +func TestSuspendActor_Finalized_CommitsDespiteObjectStoreFailure(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") - w, objects := newFinalizeWorkflow(persistence) + w, _, objects := newWireFinalizeWorkflow(t, persistence) const snapshotName = "2026-01-01t00-00-00z-new" actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, ActorTemplate: &ateapipb.ObjectRef{Atespace: "team-a", Name: "sub-tmpl"}, - Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING}, + Status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), + }, }) previous := mustActorSnapshotURI(t, template, actor, "old") @@ -639,9 +680,9 @@ func TestEnsureSuspendedFinalized_CommitsDespiteObjectStoreFailure(t *testing.T) }) objects.OnDelete = func(string, string) error { return errObjectStore } - stored, err := w.ensureSuspendedFinalized(ctx, actorRef) + stored, err := w.SuspendActor(ctx, actorRef) if err != nil { - t.Fatalf("ensureSuspendedFinalized: %v", err) + t.Fatalf("SuspendActor: %v", err) } if got := stored.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { t.Errorf("state = %v, want SUSPENDED", got) @@ -661,21 +702,27 @@ func TestEnsureSuspendedFinalized_CommitsDespiteObjectStoreFailure(t *testing.T) } } -// TestEnsureSuspendedFinalized_KeepsReplacedSnapshotOnConflict verifies that a +// TestSuspendActor_Finalized_KeepsReplacedSnapshotOnConflict verifies that a // commit lost to a concurrent update releases nothing: the actor record still // names its old snapshot, so collecting it would leave the actor pointing at // deleted objects. -func TestEnsureSuspendedFinalized_KeepsReplacedSnapshotOnConflict(t *testing.T) { +func TestSuspendActor_Finalized_KeepsReplacedSnapshotOnConflict(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") - objects := objectstoretest.New() + // The actor is seeded SUSPENDING, so the mark step skips and the first + // actor update the suspend makes is its finalize commit. + w, atelet, objects := newWireFinalizeWorkflow(t, &conflictingUpdateStore{Interface: persistence}) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, ActorTemplate: &ateapipb.ObjectRef{Atespace: "team-a", Name: "sub-tmpl"}, - Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING}, + Status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), + }, }) previous := mustActorSnapshotURI(t, template, actor, "old") @@ -690,9 +737,13 @@ func TestEnsureSuspendedFinalized_KeepsReplacedSnapshotOnConflict(t *testing.T) } }) - w := &ActorWorkflow{store: &conflictingUpdateStore{Interface: persistence}, objectStorePlugin: objectstoreplugintest.ControlClient(objects)} - if _, err := w.ensureSuspendedFinalized(ctx, actorRef); apierror.Code(err) != codes.Aborted { - t.Fatalf("ensureSuspendedFinalized = %v, want code Aborted", err) + if _, err := w.SuspendActor(ctx, actorRef); apierror.Code(err) != codes.Aborted { + t.Fatalf("SuspendActor = %v, want code Aborted", err) + } + // Every actor update conflicts, so pin that the Aborted came from finalize + // and not from an earlier step that never reached atelet. + if checkpoints, _ := atelet.calls(); checkpoints != 1 { + t.Errorf("Checkpoint calls = %d, want 1 before finalize", checkpoints) } if len(objects.Snapshot(t, previous)) == 0 { t.Error("replaced external snapshot was collected though the commit lost") @@ -709,57 +760,37 @@ func (conflictingUpdateStore) UpdateActor(context.Context, resources.ActorRef, s return nil, store.ErrVersionConflict } -// TestEnsureSuspendedFinalized_KeepsCrashFromWorkerDelete verifies that a -// crash the DeleteWorker workflow commits while ensureSuspendedFinalized is -// freeing the worker is kept: the step's update is pinned to the actor version -// it checked, so it fails Aborted instead of committing SUSPENDED over the -// crash. -func TestEnsureSuspendedFinalized_KeepsCrashFromWorkerDelete(t *testing.T) { +// TestSuspendActor_Finalized_KeepsCrashFromWorkerDelete verifies that a crash the +// DeleteWorker workflow commits while the suspend's finalize is freeing the +// worker is kept: the finalize update is pinned to the actor version it +// checked, so it fails Aborted instead of committing SUSPENDED over the crash. +func TestSuspendActor_Finalized_KeepsCrashFromWorkerDelete(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} - workerName := testWorkerUID("worker-pod-1") created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, - AssignedNode: "node1", - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - }, + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{ newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snapshot-1", someActorSnapshotURI(t, testStorageLocation, "team-a", "snapshot-1"), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), }, }, }) - if _, err := persistence.CreateWorker(ctx, &ateapipb.Worker{ - Metadata: &ateapipb.ResourceMetadata{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - NodeName: "node1", - Status: &ateapipb.WorkerStatus{}, - }); err != nil { - t.Fatalf("CreateWorker: %v", err) - } - seedAssignment(t, persistence, workerName, &ateapipb.ActorAssignment{ - Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorUid: created.GetMetadata().GetUid(), - }) + seedWireWorker(t, ctx, persistence, assignmentOf(created)) - w := &ActorWorkflow{store: workerDeletedDuringReleaseStore{Interface: persistence}} - _, err := w.ensureSuspendedFinalized(ctx, actorRef) + w, _ := newWireCaptureWorkflow(t, workerDeletedDuringReleaseStore{Interface: persistence}) + _, err := w.SuspendActor(ctx, actorRef) assertConcurrentCrashKept(t, ctx, persistence, actorRef, err, "suspend failed: "+crashMessageWorkerPodGone) } -func TestEnsureSuspendedFinalized_ReleasesOnlyOwnWorker(t *testing.T) { +func TestSuspendActor_Finalized_ReleasesOnlyOwnWorker(t *testing.T) { tests := []struct { name string assignmentAtespace string @@ -788,17 +819,15 @@ func TestEnsureSuspendedFinalized_ReleasesOnlyOwnWorker(t *testing.T) { t.Run(tt.name, func(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) + w, _ := newWireCaptureWorkflow(t, persistence) actor := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "shared"}, + Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "shared"}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: testWorkerUID("pod-1")}, - WorkerNamespace: "worker-ns", - WorkerPool: "pool", - WorkerPod: "pod-1", - WorkerPodUid: testWorkerUID("pod-1"), - }, + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{ newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snapshot-1", someActorSnapshotURI(t, testStorageLocation, "team-a", "snapshot-1"), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), @@ -811,28 +840,16 @@ func TestEnsureSuspendedFinalized_ReleasesOnlyOwnWorker(t *testing.T) { if tt.assignmentAtespace != "team-a" || tt.mismatchedUID { uid = "other-actor-uid-b" } - worker := &ateapipb.Worker{ - Metadata: &ateapipb.ResourceMetadata{Name: testWorkerUID("pod-1")}, - WorkerNamespace: "worker-ns", - WorkerPool: "pool", - WorkerPod: "pod-1", - WorkerPodUid: testWorkerUID("pod-1"), - Status: &ateapipb.WorkerStatus{}, - } - if _, err := persistence.CreateWorker(ctx, worker); err != nil { - t.Fatalf("CreateWorker: %v", err) - } - seedAssignment(t, persistence, testWorkerUID("pod-1"), &ateapipb.ActorAssignment{ + seedWireWorker(t, ctx, persistence, &ateapipb.ActorAssignment{ Actor: &ateapipb.ObjectRef{Atespace: tt.assignmentAtespace, Name: "shared"}, ActorUid: uid, }) - w := &ActorWorkflow{store: persistence} - if _, err := w.ensureSuspendedFinalized(ctx, resources.ActorRef{Atespace: "team-a", Name: "shared"}); err != nil { - t.Fatalf("ensureSuspendedFinalized: %v", err) + if _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "shared"}); err != nil { + t.Fatalf("SuspendActor: %v", err) } - stored := firstAssignment(t, persistence, testWorkerUID("pod-1")) + stored := firstAssignment(t, persistence, wireTestAssignment().GetWorker().GetName()) if released := stored == nil; released != tt.wantReleased { t.Errorf("worker released = %t, want %t (assignment: %v)", released, tt.wantReleased, stored) } @@ -901,18 +918,23 @@ func TestIsPausedOriginSuspend(t *testing.T) { } } -// TestEnsurePausedSnapshotUploaded_Preconditions covers the paused branch's -// failure handling: a lost node record crashes the actor (the snapshot can -// never be found), while an unreachable atelet stays retryable (the bytes -// are likely still on the node's disk). -func TestEnsurePausedSnapshotUploaded_Preconditions(t *testing.T) { +// TestSuspendActor_PausedSnapshotUploaded_Preconditions covers the paused +// branch's failure handling: a lost node record crashes the actor (the +// snapshot can never be found), while an unreachable atelet stays retryable +// (the bytes are likely still on the node's disk). Both actors are SUSPENDING +// with no worker, so the mark step skips and the suspend takes the paused +// branch. +func TestSuspendActor_PausedSnapshotUploaded_Preconditions(t *testing.T) { t.Run("no node recorded crashes", func(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) - w := &ActorWorkflow{store: persistence, dialer: newDanglingDialer()} + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}) + w := newMarkOnlyWorkflow(persistence) + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} - created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, LastAssignedGeneration: 1, @@ -922,29 +944,32 @@ func TestEnsurePausedSnapshotUploaded_Preconditions(t *testing.T) { }, }) - _, err := w.ensurePausedSnapshotUploaded(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, created, &ateapipb.ActorTemplate{}) + _, err := w.SuspendActor(ctx, actorRef) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code(err) = %v, want %v for a missing node record (err: %v)", got, codes.FailedPrecondition, err) } - stored, err := persistence.GetActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) - if err != nil { - t.Fatalf("GetActor: %v", err) - } + stored := mustReadActor(t, ctx, persistence, actorRef) if stored.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_CRASHED { t.Errorf("state = %v, want CRASHED", stored.GetStatus().GetState()) } + if msg, want := stored.GetStatus().GetCrash().GetMessage(), "suspend failed: "+crashMessageLocalSnapshotNodeUnknown; msg != want { + t.Errorf("crash message = %q, want %q", msg, want) + } }) t.Run("no atelet on node stays retryable", func(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) - w := &ActorWorkflow{store: persistence, dialer: newDanglingDialer()} + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}) + w := newMarkOnlyWorkflow(persistence) + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} snap := newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snap", someActorSnapshotURI(t, testStorageLocation, "team-a", "snap-dest"), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) snap.Locality = "node1" - created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, AssignedNode: "node1", @@ -953,37 +978,36 @@ func TestEnsurePausedSnapshotUploaded_Preconditions(t *testing.T) { }, }) - tmpl := &ateapipb.ActorTemplate{SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}} - _, err := w.ensurePausedSnapshotUploaded(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, created, tmpl) + _, err := w.SuspendActor(ctx, actorRef) if !errors.Is(err, ErrNoAteletOnNode) { - t.Fatalf("ensurePausedSnapshotUploaded = %v, want ErrNoAteletOnNode", err) + t.Fatalf("SuspendActor = %v, want ErrNoAteletOnNode", err) } - stored, err := persistence.GetActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) - if err != nil { - t.Fatalf("GetActor: %v", err) - } + stored := mustReadActor(t, ctx, persistence, actorRef) if stored.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { t.Errorf("state = %v, want SUSPENDING (retryable, not crashed)", stored.GetStatus().GetState()) } }) } -// TestEnsurePausedSnapshotUploaded_WireRequest verifies the PromoteSnapshot -// request names the paused actor's local snapshot, with its generation, and -// the destination the suspend recorded for it. -func TestEnsurePausedSnapshotUploaded_WireRequest(t *testing.T) { +// TestSuspendActor_PausedSnapshotUploaded_WireRequest verifies the +// PromoteSnapshot request names the paused actor's local snapshot, with its +// generation, and the destination the suspend recorded for it. +func TestSuspendActor_PausedSnapshotUploaded_WireRequest(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation, PreferredFidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}) w, atelet := newWireCaptureWorkflow(t, persistence) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} destURI := someActorSnapshotURI(t, testStorageLocation, "team-a", "snap-dest") snap := newDurableSnapshot(3, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "pause-snap", destURI, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) snap.Locality = "node-1" - created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + // A suspend from paused in progress: SUSPENDING with no worker, so the + // suspend skips its mark step and promotes the local snapshot. + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorTemplate: &ateapipb.ObjectRef{Atespace: "ns", Name: "tmpl1"}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, AssignedNode: "node-1", @@ -991,13 +1015,9 @@ func TestEnsurePausedSnapshotUploaded_WireRequest(t *testing.T) { Snapshots: []*ateapipb.Snapshot{snap}, }, }) - tmpl := &ateapipb.ActorTemplate{ - Metadata: &ateapipb.ResourceMetadata{Atespace: "ns", Name: "tmpl1"}, - SnapshotConfig: &ateapipb.SnapshotConfig{PreferredFidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY}, - } - if _, err := w.ensurePausedSnapshotUploaded(ctx, actorRef, created, tmpl); err != nil { - t.Fatalf("ensurePausedSnapshotUploaded: %v", err) + if _, err := w.SuspendActor(ctx, actorRef); err != nil { + t.Fatalf("SuspendActor: %v", err) } req := atelet.promoteRequest() if req == nil { @@ -1014,53 +1034,38 @@ func TestEnsurePausedSnapshotUploaded_WireRequest(t *testing.T) { } } -// TestEnsureSuspendedFinalized_MissingDurableSnapshotClearsAssignment verifies -// that if the latest generation snapshot entry is missing when -// ensureSuspendedFinalized runs, the actor transitions to CRASHED and clears -// WorkerAssignment after releasing the worker rather than getting stuck in -// SUSPENDING, and the step fails. -func TestEnsureSuspendedFinalized_MissingDurableSnapshotClearsAssignment(t *testing.T) { - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() +// TestSuspendActor_Finalized_MissingDurableSnapshotClearsAssignment verifies that if the +// latest generation snapshot entry is missing when the suspend finalizes, the +// actor transitions to CRASHED and clears WorkerAssignment after releasing the +// worker rather than getting stuck in SUSPENDING, and the suspend fails. +func TestSuspendActor_Finalized_MissingDurableSnapshotClearsAssignment(t *testing.T) { ctx := context.Background() + st := newTestPersistence(t) + seedSuspendTemplate(t, ctx, st, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) + w, atelet := newWireCaptureWorkflow(t, st) actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} records := crashRecords(t) - workerName := testWorkerUID("worker-pod-1") created := storetest.MustCreateActor(t, ctx, st, &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), Status: &ateapipb.ActorStatus{ - State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, - AssignedNode: "node1", - WorkerAssignment: &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - }, + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), LastAssignedGeneration: 1, + Snapshots: []*ateapipb.Snapshot{ + newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snapshot-1", someActorSnapshotURI(t, testStorageLocation, "team-a", "snapshot-1"), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), + }, }, }) - if _, err := st.CreateWorker(ctx, &ateapipb.Worker{ - Metadata: &ateapipb.ResourceMetadata{Name: workerName}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - WorkerPodUid: workerName, - NodeName: "node1", - Status: &ateapipb.WorkerStatus{}, - }); err != nil { - t.Fatalf("CreateWorker: %v", err) - } - seedAssignment(t, st, workerName, &ateapipb.ActorAssignment{ - Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name}, - ActorUid: created.GetMetadata().GetUid(), + seedWireWorker(t, ctx, st, assignmentOf(created)) + updateDuringCheckpoint(t, atelet, st, actorRef, func(s *ateapipb.ActorStatus) { + s.Snapshots = nil }) - w := &ActorWorkflow{store: st} - if _, err := w.ensureSuspendedFinalized(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { - t.Fatalf("ensureSuspendedFinalized error = %v, want FailedPrecondition for the actor it crashed", err) + if _, err := w.SuspendActor(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("SuspendActor error = %v, want FailedPrecondition for the actor it crashed", err) } got, err := st.GetActor(ctx, actorRef) if err != nil { @@ -1079,3 +1084,337 @@ func TestEnsureSuspendedFinalized_MissingDurableSnapshotClearsAssignment(t *test t.Fatalf("got %d crash records, want 1", len(*records)) } } + +// suspendModes runs a table test against both suspend workflows: sync, whose +// snapshot is durable before it returns, and async, whose snapshot stays +// node-local. +var suspendModes = []struct { + name string + run func(*ActorWorkflow, context.Context, resources.ActorRef) (*ateapipb.Actor, error) +}{ + {"sync", (*ActorWorkflow).SuspendActor}, + {"async", (*ActorWorkflow).PauseActor}, +} + +// suspendOutcome is what one suspend workflow does to a seeded actor. Rows +// omit zero-valued fields, which the matrix still pins. +type suspendOutcome struct { + code codes.Code + atelet ateletCall + state ateapipb.ActorState + assignedNode string + // workerHeld is whether the worker wireTestAssignment names still holds + // the actor's assignment. + workerHeld bool + generation int32 + snapshots []snapshotOutcome +} + +// ateletCall is the suspend RPC atelet received; the zero value is none. +type ateletCall struct { + checkpoints, promotes int + // fidelity and object describe the request sent; uploadMode is set only + // on a Checkpoint. + fidelity ateletpb.SnapshotFidelity + uploadMode ateletpb.UploadMode + object bool +} + +// snapshotOutcome is the part of a stored snapshot the matrix pins. +type snapshotOutcome struct { + generation int32 + fidelity ateapipb.SnapshotFidelity + locality string + // durable is the durable copy's status, UNSPECIFIED when there is none. + durable ateapipb.SnapshotStorageStatus +} + +// TestSuspendAndPause_CharacteristicMatrix pins, for each actor shape either +// workflow can be called on, what SuspendActor (sync) and PauseActor (async) +// do: the code returned, the atelet RPC sent, the record left behind, and +// whether the actor's worker was released. +func TestSuspendAndPause_CharacteristicMatrix(t *testing.T) { + const node = "node-1" + volumes := ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES + durable := func(gen int32, status ateapipb.SnapshotStorageStatus) *ateapipb.Snapshot { + name := fmt.Sprintf("snap-%d", gen) + return newDurableSnapshot(gen, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, volumes, "", name, someActorSnapshotURI(t, testStorageLocation, "team-a", name), status) + } + local := func(gen int32, locality string) *ateapipb.Snapshot { + return newLocalSnapshot(gen, volumes, "", fmt.Sprintf("snap-%d", gen), locality) + } + completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED + pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING + noDurable := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_UNSPECIFIED + promoting := durable(1, pending) + promoting.Locality = node + + ateletVolumes := ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES + immediate := ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE + checkpointSync := ateletCall{checkpoints: 1, fidelity: ateletVolumes, uploadMode: immediate, object: true} + checkpointAsync := ateletCall{checkpoints: 1, fidelity: ateletVolumes, uploadMode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED} + promote := ateletCall{promotes: 1, fidelity: ateletVolumes, object: true} + // rejected is a FailedPrecondition outcome: no atelet call, and the + // record left as want describes it. + rejected := func(want suspendOutcome) suspendOutcome { + want.code = codes.FailedPrecondition + return want + } + suspended := ateapipb.ActorState_ACTOR_STATE_SUSPENDED + + tests := []struct { + name string + // atespace defaults to team-a. + atespace string + status *ateapipb.ActorStatus + sync, async suspendOutcome + }{ + { + name: "RUNNING", + status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_RUNNING, AssignedNode: node, WorkerAssignment: wireTestAssignment()}, + sync: suspendOutcome{code: codes.OK, atelet: checkpointSync, state: suspended, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", completed}}}, + async: suspendOutcome{code: codes.OK, atelet: checkpointAsync, state: suspended, assignedNode: node, generation: 1, snapshots: []snapshotOutcome{{1, volumes, node, noDurable}}}, + }, + { + name: "RUNNING resumed from pause", + status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, AssignedNode: node, WorkerAssignment: wireTestAssignment(), + LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{local(1, node)}, + }, + sync: suspendOutcome{code: codes.OK, atelet: checkpointSync, state: suspended, generation: 2, snapshots: []snapshotOutcome{{2, volumes, "", completed}}}, + async: suspendOutcome{code: codes.OK, atelet: checkpointAsync, state: suspended, assignedNode: node, generation: 2, snapshots: []snapshotOutcome{{2, volumes, node, noDurable}}}, + }, + { + name: "RUNNING resumed from a durable snapshot", + status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, AssignedNode: node, WorkerAssignment: wireTestAssignment(), + LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{durable(1, completed)}, + }, + sync: suspendOutcome{code: codes.OK, atelet: checkpointSync, state: suspended, generation: 2, snapshots: []snapshotOutcome{{2, volumes, "", completed}}}, + async: suspendOutcome{code: codes.OK, atelet: checkpointAsync, state: suspended, assignedNode: node, generation: 2, snapshots: []snapshotOutcome{{1, volumes, "", completed}, {2, volumes, node, noDurable}}}, + }, + { + name: "SUSPENDED with a durable snapshot", + status: &ateapipb.ActorStatus{State: suspended, LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{durable(1, completed)}}, + sync: suspendOutcome{code: codes.OK, state: suspended, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", completed}}}, + async: rejected(suspendOutcome{state: suspended, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", completed}}}), + }, + { + name: "paused", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: node, LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{local(1, node)}}, + sync: suspendOutcome{code: codes.OK, atelet: promote, state: suspended, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", completed}}}, + async: suspendOutcome{code: codes.OK, state: suspended, assignedNode: node, generation: 1, snapshots: []snapshotOutcome{{1, volumes, node, noDurable}}}, + }, + { + name: "suspend from RUNNING in progress", + status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, AssignedNode: node, WorkerAssignment: wireTestAssignment(), + LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{durable(1, pending)}, + }, + sync: suspendOutcome{code: codes.OK, atelet: checkpointSync, state: suspended, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", completed}}}, + async: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, assignedNode: node, workerHeld: true, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", pending}}}), + }, + { + name: "suspend from paused in progress", + status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, AssignedNode: node, + LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{promoting}, + }, + sync: suspendOutcome{code: codes.OK, atelet: promote, state: suspended, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", completed}}}, + async: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, assignedNode: node, generation: 1, snapshots: []snapshotOutcome{{1, volumes, node, pending}}}), + }, + { + name: "pause in progress", + status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, AssignedNode: node, WorkerAssignment: wireTestAssignment(), + LastAssignedGeneration: 1, Snapshots: []*ateapipb.Snapshot{local(1, "")}, + }, + sync: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, assignedNode: node, workerHeld: true, generation: 1, snapshots: []snapshotOutcome{{1, volumes, "", noDurable}}}), + async: suspendOutcome{code: codes.OK, atelet: checkpointAsync, state: suspended, assignedNode: node, generation: 1, snapshots: []snapshotOutcome{{1, volumes, node, noDurable}}}, + }, + { + name: "RESUMING", + status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_RESUMING}, + sync: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_RESUMING}), + async: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_RESUMING}), + }, + { + name: "CRASHED", + status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_CRASHED}, + sync: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_CRASHED}), + async: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_CRASHED}), + }, + { + name: "DELETING", + status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_DELETING}, + sync: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_DELETING}), + async: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_DELETING}), + }, + { + // A golden actor commits MEMORY whatever the template prefers, and + // cannot be paused at all. + name: "golden RUNNING", + atespace: resources.GoldenActorAtespace, + status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_RUNNING, AssignedNode: node, WorkerAssignment: wireTestAssignment()}, + sync: suspendOutcome{code: codes.OK, atelet: ateletCall{checkpoints: 1, fidelity: ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, uploadMode: immediate, object: true}, state: suspended, generation: 1, snapshots: []snapshotOutcome{{1, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", completed}}}, + async: rejected(suspendOutcome{state: ateapipb.ActorState_ACTOR_STATE_RUNNING, assignedNode: node, workerHeld: true}), + }, + } + + for i, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + for _, mode := range suspendModes { + t.Run(mode.name, func(t *testing.T) { + want := tc.sync + if mode.name == "async" { + want = tc.async + } + atespace := tc.atespace + if atespace == "" { + atespace = "team-a" + } + // Each run gets its own store, so it can seed the one worker + // wireTestAssignment names. + ctx := context.Background() + persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation, PreferredFidelity: volumes}) + actorRef := resources.ActorRef{Atespace: atespace, Name: fmt.Sprintf("row%d-%s", i, mode.name)} + created := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), + Status: proto.CloneOf(tc.status), + }) + var assignment *ateapipb.ActorAssignment + if tc.status.GetWorkerAssignment() != nil { + assignment = assignmentOf(created) + } + seedWireWorker(t, ctx, persistence, assignment) + w, atelet := newWireCaptureWorkflow(t, persistence) + + _, err := mode.run(w, ctx, actorRef) + stored := mustReadActor(t, ctx, persistence, actorRef) + got := suspendOutcome{ + code: apierror.Code(err), + atelet: describeAteletCall(atelet), + state: stored.GetStatus().GetState(), + assignedNode: stored.GetStatus().GetAssignedNode(), + workerHeld: firstAssignment(t, persistence, wireTestAssignment().GetWorker().GetName()) != nil, + generation: stored.GetStatus().GetLastAssignedGeneration(), + snapshots: describeSnapshots(stored.GetStatus()), + } + if diff := cmp.Diff(want, got, cmp.AllowUnexported(suspendOutcome{}, ateletCall{}, snapshotOutcome{})); diff != "" { + t.Errorf("outcome mismatch (-want +got):\n%s (err: %v)", diff, err) + } + if got.state == ateapipb.ActorState_ACTOR_STATE_SUSPENDED && stored.GetStatus().GetWorkerAssignment() != nil { + t.Errorf("WorkerAssignment = %v, want cleared on SUSPENDED", stored.GetStatus().GetWorkerAssignment()) + } + assertRequestNamesLatestSnapshot(t, atelet, stored) + }) + } + }) + } +} + +// TestSuspendAndPause_FinalizeWithoutAssignment pins where the two finalize +// steps differ when the DeleteWorker workflow crashes the actor, clearing its +// worker assignment, between the atelet call and finalize. Both sides pin +// bugs tracked in #2676: the suspend commits SUSPENDED over the crash, and +// the pause keeps the crash but returns OK for an actor left CRASHED. The fix +// changes the sync wantState to CRASHED and both codes to FailedPrecondition, +// replacing the err != nil fatal below. +func TestSuspendAndPause_FinalizeWithoutAssignment(t *testing.T) { + wantState := map[string]ateapipb.ActorState{ + "sync": ateapipb.ActorState_ACTOR_STATE_SUSPENDED, + "async": ateapipb.ActorState_ACTOR_STATE_CRASHED, + } + for _, mode := range suspendModes { + t.Run(mode.name, func(t *testing.T) { + ctx := context.Background() + persistence := newTestPersistence(t) + seedSuspendTemplate(t, ctx, persistence, &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}) + w, atelet := newWireCaptureWorkflow(t, persistence) + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} + storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: suspendTestTemplateRef(), + Status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + AssignedNode: "node-1", + WorkerAssignment: wireTestAssignment(), + }, + }) + updateDuringCheckpoint(t, atelet, persistence, actorRef, func(s *ateapipb.ActorStatus) { + s.State = ateapipb.ActorState_ACTOR_STATE_CRASHED + s.WorkerAssignment = nil + }) + + returned, err := mode.run(w, ctx, actorRef) + // #2676: both modes should return FailedPrecondition here. + if err != nil { + t.Fatalf("%s: %v", mode.name, err) + } + if got := returned.GetStatus().GetState(); got != wantState[mode.name] { + t.Errorf("returned state = %v, want %v", got, wantState[mode.name]) + } + if got := mustReadActor(t, ctx, persistence, actorRef).GetStatus().GetState(); got != wantState[mode.name] { + t.Errorf("stored state = %v, want %v", got, wantState[mode.name]) + } + }) + } +} + +// describeAteletCall summarizes the suspend RPCs atelet received: how many +// of each, and the fidelity, upload mode, and object destination requested. +func describeAteletCall(atelet *capturingAtelet) ateletCall { + var call ateletCall + call.checkpoints, call.promotes = atelet.calls() + var snap *ateletpb.Snapshot + if req := atelet.checkpointRequest(); req != nil { + snap = req.GetSnapshot() + call.uploadMode = req.GetUploadConfig().GetMode() + } else if req := atelet.promoteRequest(); req != nil { + snap = req.GetSnapshot() + } + call.fidelity = snap.GetFidelity() + call.object = snap.GetObject() != nil + return call +} + +// describeSnapshots summarizes each stored snapshot as its generation, +// fidelity, locality, and durable copy's status. +func describeSnapshots(status *ateapipb.ActorStatus) []snapshotOutcome { + var out []snapshotOutcome + for _, snap := range status.GetSnapshots() { + out = append(out, snapshotOutcome{ + generation: snap.GetGeneration(), + fidelity: snap.GetFidelity(), + locality: snap.GetLocality(), + durable: snap.GetDurableSnapshot().GetStatus(), + }) + } + return out +} + +// assertRequestNamesLatestSnapshot verifies the suspend RPC atelet received, +// if any, names the snapshot at the stored actor's latest generation, and that +// an object destination is that snapshot's durable copy. +func assertRequestNamesLatestSnapshot(t *testing.T, atelet *capturingAtelet, stored *ateapipb.Actor) { + t.Helper() + var snap *ateletpb.Snapshot + if req := atelet.checkpointRequest(); req != nil { + snap = req.GetSnapshot() + } else if req := atelet.promoteRequest(); req != nil { + snap = req.GetSnapshot() + } else { + return + } + latest := snapshotAtLatestGeneration(stored.GetStatus()) + if got, want := snap.GetSnapshotUuid(), latest.GetUuid(); got != want { + t.Errorf("request Snapshot.SnapshotUuid = %q, want the latest generation's %q", got, want) + } + if obj := snap.GetObject(); obj != nil { + if got, want := obj.GetSnapshotUri(), latest.GetDurableSnapshot().GetObject().GetSnapshotUri(); got != want { + t.Errorf("request Snapshot.Object.SnapshotUri = %q, want the latest generation's durable copy %q", got, want) + } + } +} diff --git a/cmd/ateapi/internal/controlapi/workflow_test.go b/cmd/ateapi/internal/controlapi/workflow_test.go index c0e1876941..1841db374b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_test.go @@ -47,10 +47,8 @@ func TestActorStateChangeRecords(t *testing.T) { { name: "suspend marks the actor suspending", seedState: ateapipb.ActorState_ACTOR_STATE_RUNNING, - transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, actor *ateapipb.Actor, tmpl *ateapipb.ActorTemplate) { - if _, err := w.ensureMarkedSuspending(context.Background(), ref, actor, tmpl); err != nil { - t.Fatalf("ensureMarkedSuspending: %v", err) - } + transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, actor *ateapipb.Actor, _ *ateapipb.ActorTemplate) { + markOnly(t, w, ref, actor, w.SuspendActor) }, wantOp: ateattr.OperationSuspend, wantState: ateattr.ActorStateSuspending, @@ -59,10 +57,8 @@ func TestActorStateChangeRecords(t *testing.T) { { name: "pause marks the actor suspending", seedState: ateapipb.ActorState_ACTOR_STATE_RUNNING, - transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, actor *ateapipb.Actor, tmpl *ateapipb.ActorTemplate) { - if _, err := w.ensureMarkedPausing(context.Background(), ref, actor, tmpl); err != nil { - t.Fatalf("ensureMarkedPausing: %v", err) - } + transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, actor *ateapipb.Actor, _ *ateapipb.ActorTemplate) { + markOnly(t, w, ref, actor, w.PauseActor) }, wantOp: ateattr.OperationPause, wantState: ateattr.ActorStateSuspending, @@ -152,6 +148,23 @@ func TestActorStateChangeRecords(t *testing.T) { } } +// markOnly runs a suspend or pause on a RUNNING actor only as far as its mark +// step: the actor gets a worker to checkpoint on, but the atelet dial fails, +// so the SUSPENDING mark is the only transition the workflow commits. +func markOnly(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, actor *ateapipb.Actor, run func(context.Context, resources.ActorRef) (*ateapipb.Actor, error)) { + t.Helper() + ctx := context.Background() + if _, err := w.store.UpdateActor(ctx, ref, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error { + toUpdate.Status.WorkerAssignment = wireTestAssignment() + return nil + }); err != nil { + t.Fatalf("UpdateActor: %v", err) + } + w.dialer = newDanglingDialer() + _, err := run(ctx, ref) + assertStoppedAtDial(t, err) +} + // TestActorCreatedRecord covers creation. An actor created and never resumed // makes no other transition, so without this it has no record at all, at any // retention. diff --git a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go index a1363681f1..3468e5066e 100644 --- a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go @@ -23,11 +23,13 @@ import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" "github.com/agent-substrate/substrate/internal/apierror" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/objectstore/objectstoretest" "github.com/agent-substrate/substrate/internal/objectstoreplugin/objectstoreplugintest" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "github.com/google/uuid" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" "google.golang.org/grpc/codes" ) @@ -69,6 +71,17 @@ func newFinalizeWorkflow(persistence store.Interface) (*ActorWorkflow, *objectst return &ActorWorkflow{store: persistence, objectStorePlugin: objectstoreplugintest.ControlClient(objects)}, objects } +// newWireFinalizeWorkflow is newWireCaptureWorkflow with an in-memory object +// store behind the control object-store plugin, for the suspend tests that +// check which external snapshots a finalize collects. +func newWireFinalizeWorkflow(t *testing.T, persistence store.Interface) (*ActorWorkflow, *capturingAtelet, *objectstoretest.Fake) { + t.Helper() + w, atelet := newWireCaptureWorkflow(t, persistence) + objects := objectstoretest.New() + w.objectStorePlugin = objectstoreplugintest.ControlClient(objects) + return w, atelet, objects +} + // mustActorSnapshotURI builds the URI of a snapshot the actor took under // template's storage location, the way the suspend workflow does. func mustActorSnapshotURI(t *testing.T, template *ateapipb.ActorTemplate, actor *ateapipb.Actor, name string) resources.SnapshotURI { @@ -160,9 +173,9 @@ func withPausedShape(node string) func(*ateapipb.Actor) { } } -// withPauseInProgress seeds the shape a pause has between MarkPausing and -// FinalizePaused: a node-local snapshot at the latest generation that has no -// locality yet. Pair it with ACTOR_STATE_SUSPENDING. +// withPauseInProgress seeds the shape a pause has between marking the actor +// SUSPENDING and finalizing it: a node-local snapshot at the latest generation +// that has no locality yet. Pair it with ACTOR_STATE_SUSPENDING. func withPauseInProgress() func(*ateapipb.Actor) { return func(a *ateapipb.Actor) { a.Status.LastAssignedGeneration = 1 @@ -207,6 +220,121 @@ func assertPrerequisiteResult(t *testing.T, st ateapipb.ActorState, err error, w } } +// suspendTestTemplateRef names the template seedSuspendTemplate stores. +func suspendTestTemplateRef() *ateapipb.ObjectRef { + return &ateapipb.ObjectRef{Atespace: "ns", Name: "tmpl1"} +} + +// seedSuspendTemplate stores the ActorTemplate the suspend and pause workflow +// tests run their actors from, with the given snapshot config, and returns it +// as stored. +func seedSuspendTemplate(t *testing.T, ctx context.Context, st store.Interface, snapshotConfig *ateapipb.SnapshotConfig) *ateapipb.ActorTemplate { + t.Helper() + ref := suspendTestTemplateRef() + storetest.MustCreateAtespace(t, ctx, st, ref.GetAtespace()) + created, err := st.CreateActorTemplate(ctx, &ateapipb.ActorTemplate{ + Metadata: &ateapipb.ResourceMetadata{Atespace: ref.GetAtespace(), Name: ref.GetName()}, + SnapshotConfig: snapshotConfig, + SandboxConfig: &ateapipb.SandboxConfig{ + SandboxClass: "gvisor", + ConfigName: "gvisor", + }, + }) + if err != nil { + t.Fatalf("CreateActorTemplate: %v", err) + } + stored, err := st.GetActorTemplate(ctx, resources.ActorTemplateRefFromActorTemplate(created)) + if err != nil { + t.Fatalf("GetActorTemplate: %v", err) + } + return stored +} + +// seedWireWorker stores the Worker wireTestAssignment names and places +// assignment on it (none when nil), so a finalize step has a real assignment +// to release. +func seedWireWorker(t *testing.T, ctx context.Context, st store.Interface, assignment *ateapipb.ActorAssignment) { + t.Helper() + wa := wireTestAssignment() + if _, err := st.CreateWorker(ctx, &ateapipb.Worker{ + Metadata: &ateapipb.ResourceMetadata{Name: wa.GetWorker().GetName()}, + WorkerNamespace: wa.GetWorkerNamespace(), + WorkerPool: wa.GetWorkerPool(), + WorkerPod: wa.GetWorkerPod(), + WorkerPodUid: wa.GetWorkerPodUid(), + NodeName: wa.GetNodeName(), + Status: &ateapipb.WorkerStatus{}, + }); err != nil { + t.Fatalf("CreateWorker: %v", err) + } + seedAssignment(t, st, wa.GetWorker().GetName(), assignment) +} + +// assignmentOf is the worker assignment that places actor on a worker. +func assignmentOf(actor *ateapipb.Actor) *ateapipb.ActorAssignment { + return &ateapipb.ActorAssignment{ + Actor: &ateapipb.ObjectRef{Atespace: actor.GetMetadata().GetAtespace(), Name: actor.GetMetadata().GetName()}, + ActorUid: actor.GetMetadata().GetUid(), + } +} + +// newMarkOnlyWorkflow builds an ActorWorkflow whose atelet dial always fails, +// so a suspend or pause stops right after its mark step and leaves the actor +// as that step committed it. +func newMarkOnlyWorkflow(st store.Interface) *ActorWorkflow { + return &ActorWorkflow{store: st, dialer: newDanglingDialer()} +} + +// assertStoppedAtDial verifies a workflow built by newMarkOnlyWorkflow failed +// dialing atelet: it got past its mark step, and the failure is retryable. +func assertStoppedAtDial(t *testing.T, err error) { + t.Helper() + if !errors.Is(err, ErrNoAteletOnNode) { + t.Fatalf("err = %v, want ErrNoAteletOnNode from the atelet dial", err) + } +} + +// assertLifecycleFidelity verifies the one lifecycle-op datapoint reader +// collected carries want as its snapshot fidelity label. +func assertLifecycleFidelity(t *testing.T, reader *sdkmetric.ManualReader, want string) { + t.Helper() + dp := singleHistogramDP(t, reader, lifecycleOpDurationMetric) + if got, ok := attrString(dp, ateattr.SnapshotFidelityKey); !ok || got != want { + t.Errorf("lifecycle-op snapshot fidelity = %q (present %t), want %q", got, ok, want) + } +} + +// updateDuringCheckpoint makes atelet's Checkpoint mutate the stored actor +// before it returns, standing in for a writer that lands between a workflow's +// atelet call and its finalize step. +func updateDuringCheckpoint(t *testing.T, atelet *capturingAtelet, st store.Interface, actorRef resources.ActorRef, mutate func(*ateapipb.ActorStatus)) { + atelet.onCheckpoint = func(ctx context.Context) error { + actor, err := st.GetActor(ctx, actorRef) + if err != nil { + t.Errorf("GetActor during Checkpoint: %v", err) + return err + } + if _, err := st.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error { + mutate(toUpdate.Status) + return nil + }); err != nil { + t.Errorf("UpdateActor during Checkpoint: %v", err) + return err + } + return nil + } +} + +// mustReadActor reads the stored actor. +func mustReadActor(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) *ateapipb.Actor { + t.Helper() + actor, err := st.GetActor(ctx, actorRef) + if err != nil { + t.Fatalf("GetActor(%s): %v", actorRef, err) + } + return actor +} + // workerDeletedDuringReleaseStore runs the DeleteWorker workflow's release of // an actor right after a finalize step frees the actor's worker. That workflow // takes no actor lease, so its crash can land between the step's read of the