Skip to content

Update suspend and pause tests to use SuspendActor and PauseActor APIs #2688

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Dmitry Berkovich (dberkov) merged 4 commits into from
Oct 15, 2026
Merged

Update suspend and pause tests to use SuspendActor and PauseActor APIs #2688

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
48036be
Drive suspend and pause workflow tests through SuspendActor and Pause…
zoez7 Oct 14, 2026
5ef010b
Address review: pin worker release and fidelity label in suspend tests
zoez7 Oct 14, 2026
f5ce8a5
Address review: key matrix outcomes and note both #2676 pins
zoez7 Oct 15, 2026
d0df266
Address review: drive hosting pause and suspend rows through the work…
zoez7 Oct 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 32 additions & 29 deletions cmd/ateapi/internal/controlapi/hosting_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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
},
Expand All @@ -120,39 +128,34 @@ 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}},
}); err != nil {
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{
Expand Down Expand Up @@ -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) {
Expand All @@ -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")
Expand Down
67 changes: 29 additions & 38 deletions cmd/ateapi/internal/controlapi/workflow_delete_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -444,62 +445,52 @@ 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),
}
})
}

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
Expand Down
Loading
Loading