From de37ee1f5763385a8b93fa6944b04c3290c9a72a Mon Sep 17 00:00:00 2001 From: Zoe Zhao Date: Fri, 16 Oct 2026 02:50:51 +0000 Subject: [PATCH 1/2] Merge the pause workflow into the suspend workflow PauseActor and SuspendActor took the same steps and differed only in the shape of the snapshot they took. Run both through one suspend workflow parameterized by a suspend mode: a sync suspend makes the snapshot durable before returning, an async suspend (PauseActor) keeps it on the worker's node and leaves the actor pinned there. Behavior is unchanged: each mode keeps its own fast-forward, rejects a SUSPENDING actor the other mode marked, and PauseActor still accepts only RUNNING, non-golden actors. Drop workflow_pause.go. --- .../internal/controlapi/workflow_pause.go | 300 ------------------ .../controlapi/workflow_pause_test.go | 5 + .../internal/controlapi/workflow_suspend.go | 225 +++++++++---- .../internal/controlapi/workflow_test.go | 4 +- 4 files changed, 167 insertions(+), 367 deletions(-) delete mode 100644 cmd/ateapi/internal/controlapi/workflow_pause.go diff --git a/cmd/ateapi/internal/controlapi/workflow_pause.go b/cmd/ateapi/internal/controlapi/workflow_pause.go deleted file mode 100644 index 9949f76475..0000000000 --- a/cmd/ateapi/internal/controlapi/workflow_pause.go +++ /dev/null @@ -1,300 +0,0 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package controlapi - -import ( - "context" - "errors" - "fmt" - "log/slog" - "time" - - "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" - "github.com/agent-substrate/substrate/internal/apierror" - "github.com/agent-substrate/substrate/internal/ateattr" - "github.com/agent-substrate/substrate/internal/preview" - "github.com/agent-substrate/substrate/internal/proto/ateletpb" - "github.com/agent-substrate/substrate/internal/resources" - "github.com/agent-substrate/substrate/pkg/proto/ateapipb" - "go.opentelemetry.io/otel/attribute" - "google.golang.org/protobuf/proto" -) - -// PauseActor executes the workflow to pause a running actor. Idempotent: -// a re-entered workflow fast-forwards past the steps a previous attempt -// completed, deriving progress from the persisted actor alone. -func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) { - ctx = preview.ContextWithOptIn(ctx) - start := time.Now() - var actor *ateapipb.Actor - var actorTemplate *ateapipb.ActorTemplate - var wireFidelity string - // Set just before finalize; nil until then, so earlier exits label - // themselves from the record they hold. - var finalAttrs []attribute.KeyValue - - defer func() { - attrs := finalAttrs - if attrs == nil { - attrs = lifecycleOpAttrs(actor, actorTemplate, "", wireFidelity) - } - w.instruments.recordLifecycleOp(ctx, ateattr.OperationSuspend, start, err, attrs...) - }() - - leaseCtx, lease, err := w.acquireActorLease(ctx, actorRef) - if err != nil { - return nil, err - } - defer lease.Close() - - actor, actorTemplate, err = w.loadActorForPause(leaseCtx, actorRef) - if err != nil { - return nil, err - } - if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED && latestSnapshotIsLocalOnly(actor.GetStatus()) { - // Fully paused already: FinalizePaused commits SUSPENDED, the local - // snapshot's locality, and the cleared worker assignment in a single - // update, so there is nothing left to do. - // This success reports no pool, and cannot: the previous attempt - // released the worker, so the record names none (#957). - return actor, nil - } - var marked *ateapipb.Actor - if marked, err = w.ensureMarkedPausing(leaseCtx, actorRef, actor, actorTemplate); err != nil { - return nil, err - } - actor = marked - if wireFidelity, err = w.ensureAteletPaused(leaseCtx, actorRef, actor, actorTemplate); err != nil { - return nil, err - } - // TODO: There is no difference between suspend and pause for now, but we - // could optimize pause by not detaching. We would need to make sure Resume - // is idempotent. - if err = w.ensureVolumesDetached(leaseCtx, actor, actorTemplate, "DetachVolumesForPause", ateattr.OperationSuspend); err != nil { - return nil, err - } - // FinalizePaused clears the WorkerAssignment the labels read, so snapshot - // them here, as crash.go does for the crash counter. - finalAttrs = lifecycleOpAttrs(actor, actorTemplate, "", wireFidelity) - var finalized *ateapipb.Actor - if finalized, err = w.ensurePausedFinalized(leaseCtx, actorRef); err != nil { - return nil, err - } - actor = finalized - return actor, nil -} - -// loadActorForPause fetches the current actor record and its template. -func (w *ActorWorkflow) loadActorForPause(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, _ *ateapipb.ActorTemplate, err error) { - ctx, done := stepSpan(ctx, "LoadActorForPause") - defer func() { err = done(err) }() - - actor, err := w.store.GetActor(ctx, actorRef) - if err != nil { - return nil, nil, err - } - actorTemplate, err := resolveActorTemplate(ctx, w.store, actor) - if err != nil { - return nil, nil, err - } - return actor, actorTemplate, nil -} - -// ensureMarkedPausing transitions a RUNNING actor to SUSPENDING, minting the -// local snapshot name. A pause in progress is a SUSPENDING actor whose latest -// generation is a pending node-local snapshot; a suspend in progress has a -// pending durable one instead. Skips when a previous attempt already marked -// the actor; the persisted name then stays authoritative for the rest of the -// workflow. -func (w *ActorWorkflow) ensureMarkedPausing(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (_ *ateapipb.Actor, err error) { - ctx, done := stepSpan(ctx, "MarkPausing") - defer func() { err = done(err) }() - - if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDING && latestSnapshotIsPendingLocal(actor.GetStatus()) { - markSkipped(ctx, "actor already SUSPENDING for a pause") - return actor, nil - } - // The pause edge only exists from RUNNING. A SUSPENDING actor that is not - // pausing is suspending, and that suspend owns it. - if actor.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_RUNNING { - return nil, apierror.FailedPrecondition("MarkPausing prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, actor.GetStatus().GetState(), ateapipb.ActorState_ACTOR_STATE_RUNNING) - } - // By design a golden actor cannot be paused — it can only be suspended - // (committed). - if actorRef.Atespace == resources.GoldenActorAtespace { - return nil, apierror.FailedPrecondition("actors in atespace %q are golden actors, which cannot be paused", actorRef.Atespace) - } - - snapshotUUID := resources.NewSnapshotName() - storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error { - toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDING - // Increment last_assigned_generation for the new Pause request. - gen := toUpdate.Status.LastAssignedGeneration + 1 - toUpdate.Status.LastAssignedGeneration = gen - snap := newLocalSnapshot( - gen, - actorTemplate.GetSnapshotConfig().GetPreferredFidelity(), - actorTemplate.GetMetadata().GetUid(), - snapshotUUID, - "", - ) - snap.SandboxRuntime = proto.CloneOf(toUpdate.GetStatus().GetWorkerAssignment().GetSandboxRuntime()) - toUpdate.Status.Snapshots = append(toUpdate.Status.Snapshots, snap) - return nil - }) - if err != nil { - if errors.Is(err, store.ErrVersionConflict) { - return nil, apierror.Aborted("concurrent update conflict, please retry") - } - return nil, err - } - logActorStateChanged(ctx, storedActor, ateattr.OperationSuspend) - return storedActor, nil -} - -// ensureAteletPaused checkpoints the workload locally on the worker node -// under the actor's persisted snapshot name. This is the atelet reentrancy -// seam (#372): the request is keyed by the actor UID, the worker pod UID, and -// the once-minted snapshot name, so a re-entered workflow re-sends the same -// semantic request; once atelet's Checkpoint is idempotent on those keys this -// step becomes fully reentrant with no changes here. -func (w *ActorWorkflow) ensureAteletPaused(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (wireFidelity string, err error) { - ctx, done := stepSpan(ctx, "CallAteletPause") - defer func() { err = done(err) }() - - assignment := actor.GetStatus().GetWorkerAssignment() - if assignment == nil { - // Missing active worker pod reference mid-pause indicates corrupted store state. - return "", crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageWorkerAssignmentMissing) - } - - ateletConn, err := w.dialer.DialForAteletOnNode(assignment.GetNodeName()) - if err != nil { - return "", fmt.Errorf("while getting atelet conn for node %q: %w", assignment.GetNodeName(), err) - } - client := ateletpb.NewAteletClient(ateletConn) - - workloadSpec, err := workloadSpecFromActorTemplate(actorTemplate, actor, nil) - if err != nil { - return "", err - } - - // Checkpoint does not carry the sandbox config: atelet uses the version the - // actor is currently running (recorded on-node at Run/Restore) and pins it - // into the snapshot manifest. - inProgressSnap := snapshotAtLatestGeneration(actor.GetStatus()) - req := &ateletpb.CheckpointRequest{ - WorkerPodUid: assignment.GetWorkerPodUid(), - Atespace: actor.GetMetadata().GetAtespace(), - ActorName: actor.GetMetadata().GetName(), - ActorTemplateAtespace: actor.GetActorTemplate().GetAtespace(), - ActorTemplateName: actor.GetActorTemplate().GetName(), - Spec: workloadSpec, - Snapshot: &ateletpb.Snapshot{ - SnapshotUuid: inProgressSnap.GetUuid(), - Generation: inProgressSnap.GetGeneration(), - Fidelity: fidelityToAtelet(actorTemplate.GetSnapshotConfig().GetPreferredFidelity()), - }, - UploadConfig: &ateletpb.UploadConfig{ - Mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED, - }, - ActorUid: actor.GetMetadata().Uid, - } - wireFidelity = ateattr.SnapshotFidelityValue(req.GetSnapshot().GetFidelity()) - - if _, err = client.Checkpoint(ctx, req); err != nil { - return wireFidelity, handleAteletError(ctx, w.store, actorRef, ateattr.OperationSuspend, "Checkpoint", false, err) - } - return wireFidelity, nil -} - -// ensurePausedFinalized releases the actor's worker (only when it is still -// owned by this actor), records where the local snapshot lives, and commits -// SUSPENDED with the assignment cleared in a single update. The actor keeps -// its assigned node: a paused actor is a SUSPENDED actor whose latest -// snapshot is node-local, and it resumes on that node. When the worker's -// node name was lost or the latest local snapshot entry is missing, the actor -// can never be resumed from local state, so it commits CRASHED instead and -// fails the step with FailedPrecondition. It re-reads the actor first so an -// out-of-band transition (e.g. the syncer crashing the actor after its worker -// died) is not overwritten: an actor that is no longer SUSPENDING is left as -// it is and the step fails with FailedPrecondition. -func (w *ActorWorkflow) ensurePausedFinalized(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) { - ctx, done := stepSpan(ctx, "FinalizePaused") - defer func() { err = done(err) }() - - latestActor, err := w.store.GetActor(ctx, actorRef) - if err != nil { - return nil, err - } - if state := latestActor.GetStatus().GetState(); state != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { - return nil, apierror.FailedPrecondition("FinalizePaused prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_SUSPENDING) - } - - // 1. Free the worker (if it hasn't been freed yet) - if assignment := latestActor.GetStatus().GetWorkerAssignment(); assignment != nil { - if latestActor.GetStatus().GetAssignedNode() == "" { - // Without a node name we cannot record where the local snapshot lives, - // so the actor can never be resumed (the scheduler would search for a - // worker on an unknown node forever). Crash it instead of committing - // SUSPENDED with a snapshot nothing can find. - return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotNodeUnknown) - } - if snapshotAtLatestGeneration(latestActor.GetStatus()) == nil { - return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotMissing) - } - - worker, err := w.store.GetWorker(ctx, assignment.GetWorker().GetName()) - if err != nil { - if !errors.Is(err, store.ErrNotFound) { - return nil, fmt.Errorf("while getting worker for release: %w", err) - } - slog.Warn("Worker already gone during finalize pause, skipping release", "worker", assignment.GetWorkerPod()) - } else { - // Drop just this actor's assignment; any other actors the worker - // hosts keep theirs. - _, err := w.store.ReleaseActorFromWorker(ctx, worker.GetMetadata().GetName(), latestActor.GetMetadata().GetUid()) - if err != nil { - if errors.Is(err, store.ErrVersionConflict) { - return nil, apierror.Aborted("concurrent update conflict, please retry") - } - return nil, err - } - } - - // 2. Clear the actor's assignment, now that the worker is freed. The - // precondition pins the version checked above, so an out-of-band write - // since then fails the update instead of being overwritten. - storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(latestActor), func(toUpdate *ateapipb.Actor) error { - toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDED - // Record the node locality of the local checkpoint and clear older local checkpoints. - snap := snapshotAtLatestGeneration(toUpdate.Status) - snap.Locality = toUpdate.GetStatus().GetAssignedNode() - pruneSnapshots(toUpdate.Status, snap) - toUpdate.Status.WorkerAssignment = nil - return nil - }) - if err != nil { - if errors.Is(err, store.ErrVersionConflict) { - return nil, apierror.Aborted("concurrent update conflict, please retry") - } - return nil, err - } - logActorStateChanged(ctx, storedActor, ateattr.OperationSuspend) - latestActor = storedActor - } - - return latestActor, nil -} diff --git a/cmd/ateapi/internal/controlapi/workflow_pause_test.go b/cmd/ateapi/internal/controlapi/workflow_pause_test.go index 89743a9fa5..5965d224b8 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_pause_test.go @@ -17,6 +17,7 @@ package controlapi import ( "context" "fmt" + "strings" "testing" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" @@ -408,6 +409,10 @@ func TestPauseActor_MarkedPausing_StateMatrix(t *testing.T) { if got := apierror.Code(err); got != codes.FailedPrecondition { t.Errorf("state %v: code = %v, want FailedPrecondition (err: %v)", seedState, got, err) } + // Only a sync suspend accepts a paused actor, so a pause must not offer it. + if strings.Contains(err.Error(), "node-local snapshot") { + t.Errorf("state %v: err = %v, want no mention of a node-local snapshot", seedState, err) + } if got := stored.GetStatus().GetState(); got != seedState { t.Errorf("state %v: stored state = %v, want it untouched", seedState, got) } diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go index f8ae72fa36..4dfdb02d0b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend.go @@ -32,13 +32,50 @@ import ( "google.golang.org/protobuf/proto" ) +// suspendMode selects which of the two suspends the suspend workflow runs. +// Both take the same steps; the mode only decides the shape of the snapshot +// they take. +type suspendMode int + +const ( + // suspendSync makes the snapshot durable before the workflow returns: + // the actor ends SUSPENDED holding a durable snapshot (SuspendActor). + suspendSync suspendMode = iota + // suspendAsync leaves the snapshot on the worker's node, to be uploaded + // later: the actor ends SUSPENDED holding a node-local snapshot and + // pinned to that node (PauseActor). Nothing uploads it yet. + suspendAsync +) + +// uploadMode is when atelet makes the checkpoint durable. +func (m suspendMode) uploadMode() ateletpb.UploadMode { + if m == suspendAsync { + return ateletpb.UploadMode_UPLOAD_MODE_DEFERRED + } + return ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE +} + // SuspendActor executes the workflow to suspend a running or paused actor: // a running actor is checkpointed on its worker, a paused actor's node-local // snapshot is uploaded. Idempotent: a re-entered workflow fast-forwards past // the steps a previous attempt completed, deriving progress from the // persisted actor alone. -func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) { +func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error) { + ctx = preview.ContextWithOptIn(ctx) + return w.suspendActor(ctx, actorRef, suspendSync) +} + +// PauseActor executes the workflow to pause a running actor: it is +// checkpointed on its worker and the snapshot is kept on that node. +// Idempotent: a re-entered workflow fast-forwards past the steps a previous +// attempt completed, deriving progress from the persisted actor alone. +func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error) { ctx = preview.ContextWithOptIn(ctx) + return w.suspendActor(ctx, actorRef, suspendAsync) +} + +// suspendActor runs the suspend workflow behind SuspendActor and PauseActor. +func (w *ActorWorkflow) suspendActor(ctx context.Context, actorRef resources.ActorRef, mode suspendMode) (_ *ateapipb.Actor, err error) { start := time.Now() var actor *ateapipb.Actor var actorTemplate *ateapipb.ActorTemplate @@ -65,31 +102,39 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.Act if err != nil { return nil, err } - if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED && !latestSnapshotIsLocalOnly(actor.GetStatus()) { - // Fully suspended already: FinalizeSuspended commits SUSPENDED and the - // cleared worker assignment in a single update, so there is nothing - // left to do. A SUSPENDED actor whose latest snapshot is node-local is - // paused, not suspended, and goes on to upload it. This success - // reports no pool, and cannot: the previous attempt released the - // worker, so the record names none (#957). - return actor, nil + // FinalizeSuspended commits SUSPENDED and the cleared worker assignment in + // a single update, so an actor already holding the snapshot this mode + // leaves behind has nothing left to do. These successes report no pool, + // and cannot: the previous attempt released the worker, so the record + // names none (#957). + if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED { + if mode == suspendAsync && latestSnapshotIsLocalOnly(actor.GetStatus()) { + return actor, nil // fully paused + } + if mode == suspendSync && !latestSnapshotIsLocalOnly(actor.GetStatus()) { + return actor, nil // fully suspended + } + // A sync suspend of a paused actor goes on to upload its node-local + // snapshot. An async suspend of a suspended actor is rejected by MarkSuspending. } // Decided before marking: once SUSPENDING is committed, the loaded status // alone can no longer tell the two origins apart. - fromPaused := isPausedOriginSuspend(actor) + // TODO(paused-suspend): remove the paused -> suspended transition before GA. + fromPaused := mode == suspendSync && isPausedOriginSuspend(actor) var marked *ateapipb.Actor - if marked, err = w.ensureMarkedSuspending(leaseCtx, actorRef, actor, actorTemplate); err != nil { + if marked, err = w.ensureMarkedSuspending(leaseCtx, actorRef, actor, actorTemplate, mode); err != nil { return nil, err } actor = marked if fromPaused { wireFidelity, err = w.ensurePausedSnapshotUploaded(leaseCtx, actorRef, actor, actorTemplate) } else { - wireFidelity, err = w.ensureAteletSuspended(leaseCtx, actorRef, actor, actorTemplate) + wireFidelity, err = w.ensureAteletSuspended(leaseCtx, actorRef, actor, actorTemplate, mode) } if err != nil { return nil, err } + // TODO(#2734): An async suspend should skip detaching here, since the actor can resume on the same node. if err = w.ensureVolumesDetached(leaseCtx, actor, actorTemplate, "DetachVolumes", ateattr.OperationSuspend); err != nil { return nil, err } @@ -97,7 +142,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.Act // them here, as crash.go does for the crash counter. finalAttrs = lifecycleOpAttrs(actor, actorTemplate, "", wireFidelity) var finalized *ateapipb.Actor - if finalized, err = w.ensureSuspendedFinalized(leaseCtx, actorRef); err != nil { + if finalized, err = w.ensureSuspendedFinalized(leaseCtx, actorRef, mode); err != nil { return nil, err } actor = finalized @@ -121,28 +166,44 @@ func (w *ActorWorkflow) loadActorForSuspend(ctx context.Context, actorRef resour } // ensureMarkedSuspending transitions a RUNNING or paused actor to -// SUSPENDING, minting the in-progress snapshot location and recording the -// actor version the snapshot will capture. Skips when a previous attempt -// already marked the actor; the persisted location and source version then -// stay authoritative for the rest of the workflow. -func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (_ *ateapipb.Actor, err error) { +// SUSPENDING, minting the in-progress snapshot and recording the actor +// version the snapshot will capture. A sync suspend mints a pending durable +// snapshot; an async one mints a node-local snapshot with no locality yet. +// Skips when a previous attempt of the same mode already marked the actor; +// the persisted snapshot and source version then stay authoritative for the +// rest of the workflow. A mark left by the other mode is rejected: that +// workflow owns the actor until it finishes. +func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, mode suspendMode) (_ *ateapipb.Actor, err error) { ctx, done := stepSpan(ctx, "MarkSuspending") defer func() { err = done(err) }() if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDING { - if latestSnapshotIsPendingLocal(actor.GetStatus()) { - // The pause workflow owns the actor until it finishes; a suspend - // can run once the actor is paused. - return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (a pause is in progress)", actorRef) + // An async mark's latest snapshot is node-local with no locality yet, + // a sync mark's has a pending durable copy. + if latestSnapshotIsPendingLocal(actor.GetStatus()) != (mode == suspendAsync) { + inProgress := "Pause" + if mode == suspendAsync { + inProgress = "Suspend" + } + return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (a previous %s is in progress)", actorRef, inProgress) } markSkipped(ctx, "actor already SUSPENDING") return actor, nil } state := actor.GetStatus().GetState() - paused := isPausedOriginSuspend(actor) + // Only a sync suspend takes a paused actor. + paused := mode == suspendSync && isPausedOriginSuspend(actor) // TODO(paused-suspend) if state != ateapipb.ActorState_ACTOR_STATE_RUNNING && !paused { + if mode == suspendAsync { + return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_RUNNING) + } return nil, apierror.FailedPrecondition("MarkSuspending prerequisite not met for Actor: %s (got: %v, want %s or %s with a node-local snapshot)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_RUNNING, ateapipb.ActorState_ACTOR_STATE_SUSPENDED) } + // Golden actors are only suspended, never paused. + // TODO: Remove this check: the suspend workflow should not special-case golden actors. + if mode == suspendAsync && actorRef.Atespace == resources.GoldenActorAtespace { + return nil, apierror.FailedPrecondition("actors in atespace %q are golden actors, which cannot be paused", actorRef.Atespace) + } // A paused actor's durable snapshot is an upload of its local checkpoint, // so it keeps that checkpoint's name; a running actor's is a new one. var snapshotName string @@ -154,10 +215,14 @@ func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef res } // Fail here rather than at checkpoint time if the template's location - // cannot produce a usable URI: nothing has been written yet. - uri, err := newInProgressSnapshotURI(actorTemplate, actor, snapshotName) - if err != nil { - return nil, err + // cannot produce a usable URI: nothing has been written yet. An async + // suspend keeps its snapshot on the node, so it needs no URI. + var uri resources.SnapshotURI + if mode == suspendSync { + // TODO: Always populate the in-progress URI. + if uri, err = newInProgressSnapshotURI(actorTemplate, actor, snapshotName); err != nil { + return nil, err + } } storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error { toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDING @@ -176,18 +241,25 @@ func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef res return nil } // Increment last_assigned_generation for the new Suspend request and create - // a pending snapshot for that generation. + // a pending snapshot for that generation: durable for a sync suspend, + // node-local with no locality until finalize for an async one. gen := toUpdate.Status.LastAssignedGeneration + 1 toUpdate.Status.LastAssignedGeneration = gen - snap := newDurableSnapshot( - gen, - ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, - preferredFidelity(actorRef.Atespace, actorTemplate), - actorTemplate.GetMetadata().GetUid(), - uri.Name(), - uri.String(), - ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING, - ) + + fidelity := preferredFidelity(actorRef.Atespace, actorTemplate) + templateUID := actorTemplate.GetMetadata().GetUid() + snap := newLocalSnapshot(gen, fidelity, templateUID, snapshotName, "") + if mode == suspendSync { + snap = newDurableSnapshot( + gen, + ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, + fidelity, + templateUID, + uri.Name(), + uri.String(), + ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING, + ) + } snap.SandboxRuntime = proto.CloneOf(toUpdate.GetStatus().GetWorkerAssignment().GetSandboxRuntime()) toUpdate.Status.Snapshots = append(toUpdate.Status.Snapshots, snap) return nil @@ -213,6 +285,7 @@ func preferredFidelity(atespace string, tmpl *ateapipb.ActorTemplate) ateapipb.S return tmpl.GetSnapshotConfig().GetPreferredFidelity() } +// TODO(paused-suspend): remove with the paused -> suspended transition. // isPausedOriginSuspend reports whether the suspend must upload a paused // actor's node-local snapshot instead of checkpointing a running workload. // A local snapshot alone does not mean paused-origin: resume restores from it @@ -232,12 +305,14 @@ func isPausedOriginSuspend(actor *ateapipb.Actor) bool { } // ensureAteletSuspended checkpoints the workload to the actor's persisted -// in-progress snapshot location. This is the atelet reentrancy seam (#372): +// in-progress snapshot location. A sync suspend has atelet upload it before +// returning; an async suspend has atelet keep it on the node. +// This is the atelet reentrancy seam (#372): // the request is keyed by the actor UID, the worker pod UID, and the // once-minted snapshot location, so a re-entered workflow re-sends the same // semantic request; once atelet's Checkpoint is idempotent on those keys this // step becomes fully reentrant with no changes here. -func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (wireFidelity string, err error) { +func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, mode suspendMode) (wireFidelity string, err error) { ctx, done := stepSpan(ctx, "CallAteletSuspend") defer func() { err = done(err) }() @@ -263,6 +338,11 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso // into the snapshot manifest. inProgressSnap := snapshotAtLatestGeneration(actor.GetStatus()) fidelity := fidelityToAtelet(preferredFidelity(actor.GetMetadata().GetAtespace(), actorTemplate)) + // An async suspend's snapshot has no durable copy, so it stays on the node. + var object *ateletpb.ObjectStorage + if uri := inProgressSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(); uri != "" { + object = &ateletpb.ObjectStorage{SnapshotUri: uri} + } req := &ateletpb.CheckpointRequest{ WorkerPodUid: assignment.GetWorkerPodUid(), Atespace: actor.GetMetadata().GetAtespace(), @@ -274,16 +354,10 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso SnapshotUuid: inProgressSnap.GetUuid(), Generation: inProgressSnap.GetGeneration(), Fidelity: fidelity, - Object: &ateletpb.ObjectStorage{ - SnapshotUri: inProgressSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(), - }, + Object: object, }, - // FinalizeSuspended marks this snapshot COMPLETED, so it must be - // durable before Checkpoint returns. IMMEDIATE temporarily keeps no - // local copy either (see UPLOAD_MODE_IMMEDIATE), which matches - // FinalizeSuspended dropping the actor's local snapshots. UploadConfig: &ateletpb.UploadConfig{ - Mode: ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE, + Mode: mode.uploadMode(), }, ActorUid: actor.GetMetadata().Uid, } @@ -295,6 +369,7 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso return wireFidelity, nil } +// TODO(paused-suspend): remove with the paused -> suspended transition. // ensurePausedSnapshotUploaded suspends a paused actor by telling the atelet // on the node holding the local pause snapshot to upload it to the actor's // persisted in-progress snapshot location; no workload runs, so there is no @@ -311,8 +386,8 @@ func (w *ActorWorkflow) ensurePausedSnapshotUploaded(ctx context.Context, actorR } nodeName := actor.GetStatus().GetAssignedNode() if nodeName == "" { - // Without the node the snapshot can never be found (mirrors - // FinalizePaused, which crashes rather than record an unknown node). + // Without the node the snapshot can never be found (mirrors an async + // FinalizeSuspended, which crashes rather than record an unknown node). return "", crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotNodeUnknown) } @@ -363,7 +438,7 @@ func newInProgressSnapshotURI(actorTemplate *ateapipb.ActorTemplate, actor *atea // ensureVolumesDetached detaches the actor's mounted external volumes from // its worker node. Detachment is idempotent, so a re-entered workflow safely -// runs it again. spanName distinguishes the suspend and pause steps in +// runs it again. spanName distinguishes the suspend and revert steps in // traces; op labels the volume metrics. // TODO replace re-execution with a proper check on the volumes' attach state. func (w *ActorWorkflow) ensureVolumesDetached(ctx context.Context, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, spanName, op string) (err error) { @@ -380,15 +455,16 @@ func (w *ActorWorkflow) ensureVolumesDetached(ctx context.Context, actor *ateapi // ensureSuspendedFinalized releases the actor's worker (only when it is still // owned by this actor), records the in-progress snapshot as the actor's -// durable snapshot, and commits SUSPENDED with the assignment cleared in a -// single update, then releases the durable snapshot that update replaced. -// When the latest snapshot has no durable snapshot to record, it commits -// CRASHED instead and fails the step. +// durable snapshot (sync) or node-local snapshot (async), and commits +// SUSPENDED with the assignment cleared in a single update, then releases the +// durable snapshot that update replaced. +// When the latest snapshot cannot be recorded, it commits CRASHED instead and +// fails the step. // It re-reads the actor first so an out-of-band transition (e.g. the syncer // crashing the actor after its worker died) is not overwritten: an actor that // is no longer SUSPENDING is left as it is and the step fails with // FailedPrecondition. -func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) { +func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef resources.ActorRef, mode suspendMode) (_ *ateapipb.Actor, err error) { ctx, done := stepSpan(ctx, "FinalizeSuspended") defer func() { err = done(err) }() @@ -417,8 +493,19 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r if state := latestActor.GetStatus().GetState(); state != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { return nil, apierror.FailedPrecondition("FinalizeSuspended prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_SUSPENDING) } - if snapshotAtLatestGeneration(latestActor.GetStatus()).GetDurableSnapshot() == nil { - return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageDurableSnapshotMissing) + switch mode { + case suspendSync: + if snapshotAtLatestGeneration(latestActor.GetStatus()).GetDurableSnapshot() == nil { + return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageDurableSnapshotMissing) + } + case suspendAsync: + // A local snapshot on an unknown node can never be found by the scheduler. + if latestActor.GetStatus().GetAssignedNode() == "" { + return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotNodeUnknown) + } + if snapshotAtLatestGeneration(latestActor.GetStatus()) == nil { + return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageLocalSnapshotMissing) + } } // 1. Free the worker (if it hasn't been freed yet) @@ -431,7 +518,7 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r } } - // 2. Finalize the actor: record its new durable snapshot and mark it SUSPENDED. This + // 2. Finalize the actor: record its new snapshot and mark it SUSPENDED. This // must run even with no worker assignment (nothing to free), or the actor // would be left SUSPENDING forever with the workflow reporting success. // The precondition pins the version checked above, so an out-of-band write @@ -439,14 +526,22 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r t = time.Now() storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(latestActor), func(toUpdate *ateapipb.Actor) error { toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDED - inProgressSt := snapshotAtLatestGeneration(toUpdate.Status).GetDurableSnapshot() - inProgressSt.Status = ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED - // Remove all local snapshots since they are superseded by the new durable snapshot. - // TODO: This will change when we merge pause and suspend and allow for both local - // and durable snapshots to be kept. - pruneSnapshots(toUpdate.Status, nil) + snap := snapshotAtLatestGeneration(toUpdate.Status) + switch mode { + case suspendSync: + snap.GetDurableSnapshot().Status = ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED + // Remove all local snapshots since they are superseded by the new durable snapshot. + // TODO: This will change when an async suspend uploads its snapshot + // and both local and durable snapshots are kept. + pruneSnapshots(toUpdate.Status, nil) + toUpdate.Status.AssignedNode = "" + case suspendAsync: + // Record the node locality of the local checkpoint and clear older local checkpoints. + // AssignedNode is kept: a paused actor resumes on that node. + snap.Locality = toUpdate.GetStatus().GetAssignedNode() + pruneSnapshots(toUpdate.Status, snap) + } toUpdate.Status.WorkerAssignment = nil - toUpdate.Status.AssignedNode = "" return nil }) dUpdateActor = time.Since(t) diff --git a/cmd/ateapi/internal/controlapi/workflow_test.go b/cmd/ateapi/internal/controlapi/workflow_test.go index 3b42a501d2..13e0aee3a8 100644 --- a/cmd/ateapi/internal/controlapi/workflow_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_test.go @@ -81,8 +81,8 @@ func TestActorStateChangeRecords(t *testing.T) { }, }, transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, _ *ateapipb.Actor, _ *ateapipb.ActorTemplate) { - if _, err := w.ensurePausedFinalized(context.Background(), ref); err != nil { - t.Fatalf("ensurePausedFinalized: %v", err) + if _, err := w.ensureSuspendedFinalized(context.Background(), ref, suspendAsync); err != nil { + t.Fatalf("ensureSuspendedFinalized: %v", err) } }, wantOp: ateattr.OperationSuspend, From 4d8d6fa3750c31fe23b40304fa04a0a82b59b1e8 Mon Sep 17 00:00:00 2001 From: Zoe Zhao Date: Fri, 16 Oct 2026 03:24:32 +0000 Subject: [PATCH 2/2] Drive finalize pause state-change test through PauseActor --- cmd/ateapi/internal/controlapi/workflow_test.go | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/workflow_test.go b/cmd/ateapi/internal/controlapi/workflow_test.go index 13e0aee3a8..f60920fc40 100644 --- a/cmd/ateapi/internal/controlapi/workflow_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_test.go @@ -71,18 +71,13 @@ func TestActorStateChangeRecords(t *testing.T) { seedOpts: []func(*ateapipb.Actor){ withPauseInProgress(), func(a *ateapipb.Actor) { - a.Status.AssignedNode = "node1" - a.Status.WorkerAssignment = &ateapipb.WorkerAssignment{ - Worker: &ateapipb.ObjectRef{Name: "worker-gone"}, - WorkerNamespace: "default", - WorkerPool: "pool1", - WorkerPod: "worker-pod-1", - } + a.Status.AssignedNode = "node-1" + a.Status.WorkerAssignment = wireTestAssignment() }, }, transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, _ *ateapipb.Actor, _ *ateapipb.ActorTemplate) { - if _, err := w.ensureSuspendedFinalized(context.Background(), ref, suspendAsync); err != nil { - t.Fatalf("ensureSuspendedFinalized: %v", err) + if _, err := w.PauseActor(context.Background(), ref); err != nil { + t.Fatalf("PauseActor: %v", err) } }, wantOp: ateattr.OperationSuspend, @@ -131,7 +126,7 @@ func TestActorStateChangeRecords(t *testing.T) { t.Fatalf("get template: %v", err) } - w := &ActorWorkflow{store: persistence} + w, _ := newWireCaptureWorkflow(t, persistence) tt.transition(t, w, actorRef, actor, tmpl) // The logs exporter is on, so the record goes to OTLP only.