From aa1a6e1b79836c2225b97d1af81d025f6c937a11 Mon Sep 17 00:00:00 2001 From: cloudonly Date: Wed, 21 Oct 2026 01:16:40 +0000 Subject: [PATCH 1/4] Preserve committed local snapshots when reverting actors Async snapshot uploads can remain PENDING after a local checkpoint commits. Preserve that checkpoint through workload teardown and snapshot cleanup. Reject Revert on SUSPENDED actors regardless of snapshot locality. --- api/ateapi/v1/ateapi.proto | 7 +- api/ateapi/v1/ateapi_grpc.pb.go | 14 +- .../controlapi/functionaltest/actor_test.go | 193 +++++++----- .../controlapi/functionaltest/main_test.go | 61 +++- .../internal/controlapi/snapshot_helpers.go | 12 +- .../internal/controlapi/workflow_delete.go | 34 ++- .../controlapi/workflow_delete_test.go | 9 +- .../controlapi/workflow_resume_test.go | 21 +- .../internal/controlapi/workflow_revert.go | 106 +++++-- .../controlapi/workflow_revert_test.go | 278 ++++++++++++++---- .../apivalidation/zz_generated.validation.go | 28 ++ .../internal/snapshottracker/detach_test.go | 68 +++++ .../internal/snapshottracker/lifecycle.go | 20 ++ cmd/atelet/lifecycle_test.go | 68 ++++- cmd/atelet/main.go | 12 + cmd/kubectl-ate/README.md | 2 +- cmd/kubectl-ate/internal/cmd/revert_actor.go | 7 +- docs/api-guide.md | 8 +- docs/architecture.md | 2 +- .../combinedvolumes/combinedvolumes_test.go | 10 +- internal/e2e/suites/demo/demo_test.go | 58 ++-- internal/proto/ateletpb/atelet.proto | 10 + 22 files changed, 776 insertions(+), 252 deletions(-) diff --git a/api/ateapi/v1/ateapi.proto b/api/ateapi/v1/ateapi.proto index a5e777a4a6..d84302fec7 100644 --- a/api/ateapi/v1/ateapi.proto +++ b/api/ateapi/v1/ateapi.proto @@ -49,10 +49,9 @@ service Control { option (method).resource = "Actor"; } - // Revert an actor to SUSPENDED state at its durable snapshot. - // Only crashed or running actors, or suspended actors with a node-local - // snapshot, can be reverted. A suspended actor without a node-local snapshot - // is already at its durable snapshot, so there is nothing to revert. + // Revert a running or crashed actor to SUSPENDED at its latest snapshot, + // including a committed local snapshot whose upload is still pending. + // An already suspended actor cannot be reverted, regardless of snapshot locality. rpc RevertActor(RevertActorRequest) returns (RevertActorResponse) { option (method).resource = "Actor"; } diff --git a/api/ateapi/v1/ateapi_grpc.pb.go b/api/ateapi/v1/ateapi_grpc.pb.go index 4c2cadb5d9..4ff42232e1 100644 --- a/api/ateapi/v1/ateapi_grpc.pb.go +++ b/api/ateapi/v1/ateapi_grpc.pb.go @@ -95,10 +95,9 @@ type ControlClient interface { SuspendActor(ctx context.Context, in *SuspendActorRequest, opts ...grpc.CallOption) (*SuspendActorResponse, error) // Resume an actor from its latest snapshot. ResumeActor(ctx context.Context, in *ResumeActorRequest, opts ...grpc.CallOption) (*ResumeActorResponse, error) - // Revert an actor to SUSPENDED state at its durable snapshot. - // Only crashed or running actors, or suspended actors with a node-local - // snapshot, can be reverted. A suspended actor without a node-local snapshot - // is already at its durable snapshot, so there is nothing to revert. + // Revert a running or crashed actor to SUSPENDED at its latest snapshot, + // including a committed local snapshot whose upload is still pending. + // An already suspended actor cannot be reverted, regardless of snapshot locality. RevertActor(ctx context.Context, in *RevertActorRequest, opts ...grpc.CallOption) (*RevertActorResponse, error) // Delete an actor. A running actor's workload is terminated, its volumes // detached, and its worker released before the record is removed. @@ -660,10 +659,9 @@ type ControlServer interface { SuspendActor(context.Context, *SuspendActorRequest) (*SuspendActorResponse, error) // Resume an actor from its latest snapshot. ResumeActor(context.Context, *ResumeActorRequest) (*ResumeActorResponse, error) - // Revert an actor to SUSPENDED state at its durable snapshot. - // Only crashed or running actors, or suspended actors with a node-local - // snapshot, can be reverted. A suspended actor without a node-local snapshot - // is already at its durable snapshot, so there is nothing to revert. + // Revert a running or crashed actor to SUSPENDED at its latest snapshot, + // including a committed local snapshot whose upload is still pending. + // An already suspended actor cannot be reverted, regardless of snapshot locality. RevertActor(context.Context, *RevertActorRequest) (*RevertActorResponse, error) // Delete an actor. A running actor's workload is terminated, its volumes // detached, and its worker released before the record is removed. diff --git a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go index 7544c3bd2a..e6e829a9a7 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go +++ b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go @@ -6494,6 +6494,14 @@ func TestRevertActor_UnreachableNode(t *testing.T) { t.Fatalf("suspended actor has no retained checkpoint: %v", suspended.GetActor().GetStatus()) } waitForWorkerAvailable(t, tc, worker) + // Recover a crashed actor with no live workload left on its old node. + stored := suspended.GetActor() + if _, err := tc.persistence.UpdateActor(ctx, resources.ActorRef{Atespace: testAtespace, Name: ref.GetName()}, store.PreconditionFrom(stored), func(a *ateapi.Actor) error { + a.Status.State = ateapi.ActorState_ACTOR_STATE_CRASHED + return nil + }); err != nil { + t.Fatal(err) + } deleteWorkerPod(t, tc, ns, "worker-1") if tt.missing { if err := tc.k8sClient.CoreV1().Pods(ateletNamespace).Delete(ctx, sourceAtelet, metav1.DeleteOptions{GracePeriodSeconds: ptr.To[int64](0)}); err != nil { @@ -6552,82 +6560,129 @@ func TestRevertActor_UnreachableNode(t *testing.T) { } } -// TestRevertActor_FromPaused reverts a suspended actor with a local snapshot -// back to its external snapshot, discarding the local snapshot info on the -// actor record. -func TestRevertActor_FromPaused(t *testing.T) { - ns := namespaceForTest("ns-revert-paused") - tc := setupTest(t, ns) - defer tc.cleanup() +// Revert must restore the latest async suspend, not the older durable fallback. +func TestRevertActor_AsyncSnapshot(t *testing.T) { + for _, cleanupFailure := range []bool{false, true} { + t.Run(fmt.Sprintf("checkpoint cleanup failure=%v", cleanupFailure), func(t *testing.T) { + ns := namespaceForTest("ns-revert-paused") + tc := setupTest(t, ns) + defer tc.cleanup() - createTemplate(t, tc, ns) - workerName := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + createTemplate(t, tc, ns) + workerName := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + tc.fakeAtelet.Lock.Lock() + tc.fakeAtelet.KeepCheckpointLocal = true + tc.fakeAtelet.Lock.Unlock() - ctx := context.Background() - const name = "id1" - actorRef := &ateapi.ObjectRef{Atespace: testAtespace, Name: name} + ctx := context.Background() + const name = "id1" + actorRef := &ateapi.ObjectRef{Atespace: testAtespace, Name: name} - if _, err := tc.client.CreateActor(ctx, &ateapi.CreateActorRequest{Actor: &ateapi.Actor{ - Metadata: &ateapi.ResourceMetadata{Atespace: testAtespace, Name: name}, - ActorTemplate: &ateapi.ObjectRef{Atespace: testAtespace, Name: "tmpl1"}, - }}); err != nil { - t.Fatalf("CreateActor failed: %v", err) - } + if _, err := tc.client.CreateActor(ctx, &ateapi.CreateActorRequest{Actor: &ateapi.Actor{ + Metadata: &ateapi.ResourceMetadata{Atespace: testAtespace, Name: name}, + ActorTemplate: &ateapi.ObjectRef{Atespace: testAtespace, Name: "tmpl1"}, + }}); err != nil { + t.Fatalf("CreateActor failed: %v", err) + } - if _, err := tc.client.ResumeActor(ctx, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { - t.Fatalf("ResumeActor failed: %v", err) - } - suspended, err := tc.client.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: actorRef, Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_SYNCHRONOUS}}) - if err != nil { - t.Fatalf("SuspendActor failed: %v", err) - } - waitForWorkerAvailable(t, tc, workerName) - snapshotURI := durableSnapshotURI(suspended.GetActor().GetStatus()) - if snapshotURI == "" { - t.Fatalf("SuspendActor wrote no external snapshot: %v", suspended) - } + if _, err := tc.client.ResumeActor(ctx, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { + t.Fatalf("ResumeActor failed: %v", err) + } + suspended, err := tc.client.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: actorRef, Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_SYNCHRONOUS}}) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + waitForWorkerAvailable(t, tc, workerName) + snapshotURI := durableSnapshotURI(suspended.GetActor().GetStatus()) + if snapshotURI == "" { + t.Fatalf("SuspendActor wrote no external snapshot: %v", suspended) + } - if _, err := tc.client.ResumeActor(ctx, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { - t.Fatalf("second ResumeActor failed: %v", err) - } - if _, err := tc.client.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: actorRef, Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_ASYNCHRONOUS}}); err != nil { - t.Fatalf("SuspendActor failed: %v", err) - } - waitForWorkerAvailable(t, tc, workerName) - tc.fakeAtelet.Lock.Lock() - tc.fakeAtelet.TerminateCalled = false - tc.fakeAtelet.CheckpointCalled = false - tc.fakeAtelet.Lock.Unlock() + if _, err := tc.client.ResumeActor(ctx, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { + t.Fatalf("second ResumeActor failed: %v", err) + } + paused, err := tc.client.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: actorRef, Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_ASYNCHRONOUS}}) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + waitForWorkerAvailable(t, tc, workerName) + wantLocal := proto.CloneOf(localSnapshot(paused.GetActor().GetStatus())) + wantGeneration := suspended.GetActor().GetStatus().GetLastAssignedGeneration() + 1 + if wantLocal.GetGeneration() != wantGeneration || wantLocal.GetDurableSnapshot().GetStatus() != ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING { + t.Fatalf("async suspend did not commit a pending local snapshot at generation %d: %v", wantGeneration, paused) + } + if _, err := tc.client.RevertActor(ctx, &ateapi.RevertActorRequest{Actor: actorRef}); status.Code(err) != codes.FailedPrecondition { + t.Fatalf("RevertActor on local SUSPENDED actor = %v, want FailedPrecondition", err) + } + if _, err := tc.client.ResumeActor(ctx, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { + t.Fatalf("ResumeActor from async snapshot: %v", err) + } + if cleanupFailure { + tc.fakeAtelet.Lock.Lock() + tc.fakeAtelet.FailCheckpointCleanup = status.Error(codes.Internal, "checkpoint committed but cleanup failed") + tc.fakeAtelet.Lock.Unlock() + if _, err := tc.client.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: actorRef, Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_ASYNCHRONOUS}}); status.Code(err) != codes.Internal { + t.Fatalf("SuspendActor after committing checkpoint = %v, want Internal", err) + } + crashed, err := tc.client.GetActor(ctx, &ateapi.GetActorRequest{Actor: actorRef}) + if err != nil { + t.Fatal(err) + } + if crashed.GetStatus().GetState() != ateapi.ActorState_ACTOR_STATE_CRASHED { + t.Fatalf("failed suspend state = %v, want CRASHED", crashed.GetStatus()) + } + for _, snap := range crashed.GetStatus().GetSnapshots() { + if snap.GetGeneration() == crashed.GetStatus().GetLastAssignedGeneration() { + if snap.GetLocality() != "" { + t.Fatal("failed suspend unexpectedly finalized locality") + } + wantLocal = proto.CloneOf(snap) + wantLocal.Locality = "node1" + } + } + } + tc.fakeAtelet.Lock.Lock() + tc.fakeAtelet.TerminateCalled = false + tc.fakeAtelet.CheckpointCalled = false + tc.fakeAtelet.Lock.Unlock() - reverted, err := tc.client.RevertActor(ctx, &ateapi.RevertActorRequest{Actor: actorRef}) - if err != nil { - t.Fatalf("RevertActor failed: %v", err) - } + reverted, err := tc.client.RevertActor(ctx, &ateapi.RevertActorRequest{Actor: actorRef}) + if err != nil { + t.Fatalf("RevertActor failed: %v", err) + } - got := reverted.GetActor().GetStatus() - if got.GetState() != ateapi.ActorState_ACTOR_STATE_SUSPENDED { - t.Errorf("state = %v, want SUSPENDED", got.GetState()) - } - if uri := durableSnapshotURI(got); uri != snapshotURI { - t.Errorf("external snapshot = %q, want it untouched at %q", uri, snapshotURI) - } - assertSnapshotPresent(t, tc, snapshotURI) - if ls := localSnapshot(got); ls != nil { - t.Errorf("local snapshot info = %v, want nil", ls) - } - if got.GetWorkerAssignment() != nil { - t.Errorf("worker assignment = %v, want nil", got.GetWorkerAssignment()) - } - if got.GetAssignedNode() != "" { - t.Errorf("assigned node = %q, want empty", got.GetAssignedNode()) - } - if !tc.fakeAtelet.TerminateCalled { - t.Errorf("expected Terminate call to clean up assigned node for paused actor") - } else if gotUID := tc.fakeAtelet.TerminateRequest.GetWorkerPodUid(); gotUID != "" { - t.Errorf("TerminateRequest.WorkerPodUid = %q, want empty for paused actor", gotUID) - } - if tc.fakeAtelet.CheckpointCalled { - t.Errorf("RevertActor checkpointed the workload, want the execution discarded") + got := reverted.GetActor().GetStatus() + if got.GetState() != ateapi.ActorState_ACTOR_STATE_SUSPENDED { + t.Errorf("state = %v, want SUSPENDED", got.GetState()) + } + if uri := durableSnapshotURI(got); uri != snapshotURI { + t.Errorf("external snapshot = %q, want it untouched at %q", uri, snapshotURI) + } + assertSnapshotPresent(t, tc, snapshotURI) + if ls := localSnapshot(got); !proto.Equal(ls, wantLocal) { + t.Errorf("local snapshot = %v, want %v", ls, wantLocal) + } + if got.GetWorkerAssignment() != nil { + t.Errorf("worker assignment = %v, want nil", got.GetWorkerAssignment()) + } + if got.GetAssignedNode() != "node1" { + t.Errorf("assigned node = %q, want node1", got.GetAssignedNode()) + } + if !tc.fakeAtelet.TerminateCalled { + t.Errorf("expected Terminate call to stop the resumed workload") + } else if req := tc.fakeAtelet.TerminateRequest; !req.GetPreserveSnapshots() { + t.Errorf("TerminateRequest = %v, want snapshots preserved", req) + } + if tc.fakeAtelet.CheckpointCalled { + t.Errorf("RevertActor checkpointed the workload, want the execution discarded") + } + if _, err := tc.client.ResumeActor(ctx, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { + t.Fatalf("ResumeActor after revert: %v", err) + } + if restored := tc.fakeAtelet.lastRestoreRequest().GetSnapshot(); restored.GetSnapshotUuid() != wantLocal.GetUuid() || restored.GetGeneration() != wantLocal.GetGeneration() { + t.Errorf("restored snapshot = %v, want retained local snapshot %v", restored, wantLocal) + } + }) } } diff --git a/cmd/ateapi/internal/controlapi/functionaltest/main_test.go b/cmd/ateapi/internal/controlapi/functionaltest/main_test.go index 12b651eee7..c7c5d2573e 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/main_test.go +++ b/cmd/ateapi/internal/controlapi/functionaltest/main_test.go @@ -122,12 +122,14 @@ type FakeAteletServer struct { RunRequest *ateletpb.RunRequest FailRun error - CheckpointCalled bool - CheckpointRequest *ateletpb.CheckpointRequest - FailCheckpoint error - KeepCheckpointLocal bool + CheckpointCalled bool + CheckpointRequest *ateletpb.CheckpointRequest + FailCheckpoint error + FailCheckpointCleanup error + KeepCheckpointLocal bool // checkpointed is each actor's last checkpointed snapshot, by UID. - checkpointed map[string]*ateletpb.Snapshot + checkpointed map[string]*ateletpb.Snapshot + snapshotStates map[string]*ateletpb.ActorSnapshotState RestoreCalled bool RestoreRequest *ateletpb.RestoreRequest @@ -187,7 +189,9 @@ func (f *FakeAteletServer) Reset() { f.CheckpointCalled = false f.CheckpointRequest = nil f.FailCheckpoint = nil + f.FailCheckpointCleanup = nil f.checkpointed = nil + f.snapshotStates = nil f.KeepCheckpointLocal = false f.RestoreCalled = false @@ -225,10 +229,15 @@ func (f *FakeAteletServer) PromoteSnapshot(ctx context.Context, req *ateletpb.Pr } // Like atelet, report the snapshot durable and still kept on the node. report := &ateletpb.SnapshotReport{Snapshot: proto.CloneOf(req.GetSnapshot())} - return &ateletpb.PromoteSnapshotResponse{State: &ateletpb.ActorSnapshotState{ + state := &ateletpb.ActorSnapshotState{ LatestLocal: report, LatestDurable: proto.CloneOf(report), - }}, nil + } + if f.snapshotStates == nil { + f.snapshotStates = map[string]*ateletpb.ActorSnapshotState{} + } + f.snapshotStates[req.GetActorUid()] = proto.CloneOf(state) + return &ateletpb.PromoteSnapshotResponse{State: state}, nil } // DetachActor uploads the actor's last checkpointed snapshot and reports it @@ -249,9 +258,14 @@ func (f *FakeAteletServer) DetachActor(ctx context.Context, req *ateletpb.Detach if err := f.writeSnapshot(snap.GetObject().GetSnapshotUri()); err != nil { return nil, err } - return &ateletpb.DetachActorResponse{State: &ateletpb.ActorSnapshotState{ + state := &ateletpb.ActorSnapshotState{ LatestDurable: &ateletpb.SnapshotReport{Snapshot: proto.CloneOf(snap)}, - }}, nil + } + if f.snapshotStates == nil { + f.snapshotStates = map[string]*ateletpb.ActorSnapshotState{} + } + f.snapshotStates[req.GetActorUid()] = proto.CloneOf(state) + return &ateletpb.DetachActorResponse{State: state}, nil } func (f *FakeAteletServer) Run(ctx context.Context, req *ateletpb.RunRequest) (*ateletpb.RunResponse, error) { @@ -289,15 +303,22 @@ func (f *FakeAteletServer) Checkpoint(ctx context.Context, req *ateletpb.Checkpo return nil, err } } + report := &ateletpb.SnapshotReport{Snapshot: proto.CloneOf(req.GetSnapshot())} + state := &ateletpb.ActorSnapshotState{} if f.KeepCheckpointLocal { - report := &ateletpb.SnapshotReport{Snapshot: proto.CloneOf(req.GetSnapshot())} - state := &ateletpb.ActorSnapshotState{LatestLocal: report} - if req.GetUploadConfig().GetMode() == ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE { - state.LatestDurable = proto.CloneOf(report) - } - return &ateletpb.CheckpointResponse{State: state}, nil + state.LatestLocal = report } - return &ateletpb.CheckpointResponse{}, nil + if req.GetUploadConfig().GetMode() == ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE { + state.LatestDurable = proto.CloneOf(report) + } + if f.snapshotStates == nil { + f.snapshotStates = map[string]*ateletpb.ActorSnapshotState{} + } + f.snapshotStates[req.GetActorUid()] = proto.CloneOf(state) + if f.FailCheckpointCleanup != nil { + return nil, f.FailCheckpointCleanup + } + return &ateletpb.CheckpointResponse{State: state}, nil } func (f *FakeAteletServer) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (*ateletpb.RestoreResponse, error) { @@ -327,6 +348,14 @@ func (f *FakeAteletServer) Terminate(ctx context.Context, req *ateletpb.Terminat if f.FailTerminate != nil { return nil, f.FailTerminate } + if req.GetPreserveSnapshots() { + state := proto.CloneOf(f.snapshotStates[req.GetActorUid()]) + if state == nil { + state = &ateletpb.ActorSnapshotState{} + } + return &ateletpb.TerminateResponse{State: state}, nil + } + delete(f.snapshotStates, req.GetActorUid()) return &ateletpb.TerminateResponse{}, nil } diff --git a/cmd/ateapi/internal/controlapi/snapshot_helpers.go b/cmd/ateapi/internal/controlapi/snapshot_helpers.go index 3e29b78312..2eb7bd3cfd 100644 --- a/cmd/ateapi/internal/controlapi/snapshot_helpers.go +++ b/cmd/ateapi/internal/controlapi/snapshot_helpers.go @@ -84,9 +84,9 @@ func clearLocalSnapshots(status *ateapi.ActorStatus) { // pruneSnapshots retains only local, the Snapshot whose local checkpoint the // actor keeps (nil if it keeps none), and the latest COMPLETED durable // snapshot, dropping all other (stale PENDING or superseded COMPLETED) -// Snapshots. It clears Locality on every Snapshot other than local: it runs -// in finalize steps that follow an atelet call that deleted the actor's other -// local checkpoints from the node. Any older COMPLETED durable snapshots are +// Snapshots. It clears Locality on every Snapshot other than local. Atelet +// retains the latest local checkpoint and any older copy an upload still needs. +// Any older COMPLETED durable snapshots are // expected to have already been garbage collected (or captured for cleanup) // beforehand. // @@ -95,8 +95,10 @@ func clearLocalSnapshots(status *ateapi.ActorStatus) { // PENDING copy. An earlier asynchronous suspend's PENDING copy is dropped: // whatever atelet uploaded of it stays under the actor's prefix until // DeleteActor collects it. -// - On revert finalization, ensureInProgressSnapshotDiscarded has already -// deleted the objects of every PENDING durable snapshot. +// - On revert finalization, the latest local snapshot and its PENDING copy +// are kept. ensureInProgressSnapshotDiscarded cleaned up incomplete +// checkpoints without locality. Older local uploads may finish under the +// actor's prefix until DeleteActor collects them. // - On synchronous suspend finalization, the latest snapshot has already been // marked COMPLETED. // diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index 3cf41f8899..d301fad43e 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -61,7 +61,7 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto } var atletTerminatedErr, volumesDetachedErr error - if err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete); err != nil { + if _, _, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete, false); err != nil { atletTerminatedErr = fmt.Errorf("while terminating atelet: %w", err) errs = append(errs, atletTerminatedErr) } @@ -97,14 +97,16 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto } // ensureAteletTerminated calls atelet to terminate the workload for opName. -func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate, opName string) (err error) { +// It reports whether the assigned node is gone, so callers do not retain local +// snapshots on a node that no longer exists. +func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate, opName string, preserveSnapshots bool) (snapshotState *ateletpb.ActorSnapshotState, nodeGone bool, err error) { ctx, done := stepSpan(ctx, "CallAteletTerminate") defer func() { err = done(err) }() nodeName := actor.GetStatus().GetAssignedNode() if nodeName == "" { slog.InfoContext(ctx, "actor has no assigned node, skipping atelet terminate request", slog.Any("actor", actorRef)) - return nil + return nil, false, nil } workerPodUID := "" @@ -116,7 +118,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res // ordinary case, and the others are none of this delete's business. hosted, err := workerHostsActor(ctx, w.store, workerName, actor.GetMetadata().GetUid()) if err != nil { - return err + return nil, false, err } if !hosted { slog.InfoContext(ctx, "worker is no longer assigned to this actor, skipping worker workload termination", @@ -131,18 +133,18 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res if errors.Is(err, ErrNoAteletOnNode) { gone, nodeErr := w.dialer.NodeGone(ctx, nodeName) if nodeErr != nil { - return fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr) + return nil, false, fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr) } if gone { // A deleted Node object counts as the node being gone for good. If the // machine still runs, whatever atelet holds for the actor stays there. slog.WarnContext(ctx, "assigned node no longer exists, skipping atelet terminate request", slog.String("node", nodeName), slog.Any("actor", actorRef)) - return nil + return nil, true, nil } } if err != nil { - return fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err) + return nil, false, fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err) } client := ateletpb.NewAteletClient(conn) @@ -151,7 +153,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res if actorTemplate != nil { spec, err := workloadSpecFromActorTemplate(actorTemplate, actor, nil) if err != nil { - return err + return nil, false, err } workloadSpec = spec } else { @@ -172,17 +174,21 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res ActorTemplateAtespace: actor.GetActorTemplate().GetAtespace(), ActorTemplateName: actor.GetActorTemplate().GetName(), Spec: workloadSpec, + PreserveSnapshots: preserveSnapshots, } - if _, err := client.Terminate(ctx, req); err != nil { - if status.Code(err) == codes.NotFound { + resp, err := client.Terminate(ctx, req) + if err != nil { + if status.Code(err) == codes.NotFound && !preserveSnapshots { slog.InfoContext(ctx, "workload already terminated on atelet", slog.Any("actor", actorRef)) - return nil + return nil, false, nil } - return handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err) + return nil, false, handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err) } - - return nil + if preserveSnapshots && resp.GetState() == nil { + return nil, false, apierror.Internal("atelet terminated Actor %s without reporting retained snapshot state", actorRef) + } + return resp.GetState(), false, nil } // fallbackWorkloadSpec is the spec to terminate an actor with when its template diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index 5eb4abbba4..1f083f7caa 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -124,7 +124,7 @@ func TestEnsureAteletTerminated_NodeGone(t *testing.T) { w := newTestActorWorkflow(t, newTestPersistence(t), "team-a", "tmpl") w.dialer = NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "", WithNodeGetter(nodeGetter(tc.lookupErr, tc.nodes...))) - err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert") + _, gone, err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert", true) if tc.wantErr != (err != nil) { t.Fatalf("ensureAteletTerminated = %v, want error %v", err, tc.wantErr) } @@ -134,6 +134,9 @@ func TestEnsureAteletTerminated_NodeGone(t *testing.T) { if tc.lookupErr != nil && !strings.Contains(err.Error(), tc.lookupErr.Error()) { t.Fatalf("ensureAteletTerminated = %v, want the node lookup error in it", err) } + if gone != !tc.wantErr { + t.Errorf("node gone = %v, want %v", gone, !tc.wantErr) + } }) } } @@ -165,9 +168,9 @@ func TestRevertAndDelete_AssignedNodeGone(t *testing.T) { // wantDetachErr expects the plugin's detach error back. wantDetachErr bool }{ - {name: "paused, node gone", state: ateapi.ActorState_ACTOR_STATE_SUSPENDED, paused: true}, + {name: "local snapshot, node gone", state: ateapi.ActorState_ACTOR_STATE_CRASHED, paused: true}, {name: "crashed, node gone", state: ateapi.ActorState_ACTOR_STATE_CRASHED}, - {name: "paused, node kept", state: ateapi.ActorState_ACTOR_STATE_SUSPENDED, paused: true, nodeKept: true, wantErr: ErrNoAteletOnNode}, + {name: "local snapshot, node kept", state: ateapi.ActorState_ACTOR_STATE_CRASHED, paused: true, nodeKept: true, wantErr: ErrNoAteletOnNode}, {name: "crashed, node kept", state: ateapi.ActorState_ACTOR_STATE_CRASHED, nodeKept: true, wantErr: ErrNoAteletOnNode}, {name: "node gone, volume detach OK", state: ateapi.ActorState_ACTOR_STATE_CRASHED, volume: true}, {name: "node gone, volume detach NotFound", state: ateapi.ActorState_ACTOR_STATE_CRASHED, volume: true, detachErr: status.Error(codes.NotFound, "no such node")}, diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index eb2e4a8072..a4ca4fd940 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -1203,6 +1203,7 @@ type capturingAtelet struct { run *ateletpb.RunRequest checkpoint *ateletpb.CheckpointRequest promote *ateletpb.PromoteSnapshotRequest + terminate *ateletpb.TerminateRequest checkpoints int promotes int @@ -1225,9 +1226,10 @@ type capturingAtelet struct { detachState *ateletpb.ActorSnapshotState detachErr error // onDetach, when set, runs before DetachActor returns. - onDetach func() - terminateErr error - onTerminate func() + onDetach func() + terminateErr error + onTerminate func() + terminateState *ateletpb.ActorSnapshotState } func (f *capturingAtelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (*ateletpb.RestoreResponse, error) { @@ -1298,12 +1300,23 @@ func (f *capturingAtelet) Terminate(ctx context.Context, req *ateletpb.Terminate hook() } f.mu.Lock() + f.terminate = proto.CloneOf(req) err := f.terminateErr + state := proto.CloneOf(f.terminateState) f.mu.Unlock() if err != nil { return nil, err } - return &ateletpb.TerminateResponse{}, nil + if req.GetPreserveSnapshots() && state == nil { + state = &ateletpb.ActorSnapshotState{} + } + return &ateletpb.TerminateResponse{State: state}, nil +} + +func (f *capturingAtelet) terminateRequest() *ateletpb.TerminateRequest { + f.mu.Lock() + defer f.mu.Unlock() + return proto.CloneOf(f.terminate) } func (f *capturingAtelet) detachRequest() *ateletpb.DetachActorRequest { diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go index 6c553fcd91..057c29ef74 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert.go @@ -27,30 +27,26 @@ import ( "go.ate.dev/substrate/internal/apierror" "go.ate.dev/substrate/internal/ateattr" "go.ate.dev/substrate/internal/preview" + "go.ate.dev/substrate/internal/proto/ateletpb" "go.ate.dev/substrate/internal/resources" "go.opentelemetry.io/otel/attribute" ) -// revertableStates are the states a revert is accepted from. A SUSPENDED -// actor is also accepted when its latest snapshot is node-local: the revert -// discards the node-local copy and returns the actor to its latest durable -// snapshot, which is that same snapshot once atelet's upload of it is -// recorded. See isRevertable. +// revertableStates are the states a revert is accepted from. Snapshot locality +// does not change which lifecycle operations an actor accepts. var revertableStates = []ateapi.ActorState{ ateapi.ActorState_ACTOR_STATE_RUNNING, ateapi.ActorState_ACTOR_STATE_CRASHED, } -// isRevertable reports whether a revert is accepted from the actor's current -// state and snapshot shape. +// isRevertable reports whether a revert is accepted from the actor's current state. func isRevertable(actor *ateapi.Actor) bool { state := actor.GetStatus().GetState() - return slices.Contains(revertableStates, state) || - (state == ateapi.ActorState_ACTOR_STATE_SUSPENDED && latestSnapshotIsLocal(actor.GetStatus())) + return slices.Contains(revertableStates, state) } // RevertActor executes the workflow to discard an actor's current execution and -// return it to SUSPENDED, leaving its external snapshot untouched for a later +// return it to SUSPENDED, leaving its latest committed snapshot untouched for a later // resume to restore from. It is the recovery verb: it accepts a CRASHED actor, // and needs no live worker to succeed. // @@ -94,14 +90,15 @@ func (w *ActorWorkflow) RevertActor(ctx context.Context, actorRef resources.Acto if err = w.ensureWorkerDiscarded(leaseCtx, actorRef, actor, actorTemplate); err != nil { return nil, err } + // Termination may have discovered a deleted node and cleared its local copies. + if actor, err = w.store.GetActor(leaseCtx, actorRef); err != nil { + return nil, err + } if err = w.ensureInProgressSnapshotDiscarded(leaseCtx, actor); err != nil { return nil, err } - // TODO: Check if local checkpoints need to be pruned #641 - // w.ensureLocalCheckpointsPruned(leaseCtx, actor) - // FinalizeReverted clears the WorkerAssignment the labels read, so snapshot // them here, as suspend and crash.go do. finalAttrs = lifecycleOpAttrs(actor, "", "") @@ -152,7 +149,7 @@ func (w *ActorWorkflow) ensureMarkedReverting(ctx context.Context, actorRef reso return actor, nil } if !isRevertable(actor) { - return nil, apierror.FailedPrecondition("Actor %s is not in a revertable state (got: %v, want one of %v, or %s with a node-local snapshot)", actorRef, st, revertableStates, ateapi.ActorState_ACTOR_STATE_SUSPENDED) + return nil, apierror.FailedPrecondition("Actor %s is not in a revertable state (got: %v, want one of %v)", actorRef, st, revertableStates) } storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapi.Actor) error { @@ -187,7 +184,8 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso if actor.GetStatus().GetAssignedNode() != "" { forgetNode := false - if err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert); err != nil { + snapshotState, nodeGone, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert, true) + if err != nil { if !canDetachWithoutAtelet(ctx, actor, err) { return err } @@ -201,6 +199,19 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso return err } } + if nodeGone || snapshotState != nil { + // A checkpoint may commit before Suspend fails, or detach while the + // actor is crashed. Reconcile before deleting any pending objects. + if _, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(a *ateapi.Actor) error { + if nodeGone { + clearLocalSnapshots(a.Status) + return nil + } + return applyRevertedSnapshotState(ctx, a.Status, actor.GetStatus().GetAssignedNode(), snapshotState) + }); err != nil { + return err + } + } if actor.GetStatus().GetWorkerAssignment() != nil { if _, err := releaseWorker(ctx, w.store, actor); err != nil { return fmt.Errorf("while releasing worker: %w", err) @@ -211,14 +222,59 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso return w.releaseAssignmentWithoutBacklink(ctx, actor) } -// ensureInProgressSnapshotDiscarded removes pending uploads of the discarded execution. -// The actor keeps its latest completed durable snapshot. +// applyRevertedSnapshotState reconciles the checkpoints atelet actually kept +// after teardown, including a checkpoint whose Suspend failed after committing. +func applyRevertedSnapshotState(ctx context.Context, status *ateapi.ActorStatus, node string, state *ateletpb.ActorSnapshotState) error { + knownLocal := latestSnapshot(status, func(s *ateapi.Snapshot) bool { return s.GetLocality() == node }) + durable := findLatestDurableSnapshot(status, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED) + committed := max(state.GetLatestLocal().GetSnapshot().GetGeneration(), state.GetLatestDurable().GetSnapshot().GetGeneration(), durable.GetGeneration()) + if knownLocal.GetGeneration() > committed { + // Tracker state can be lost after a failed save and an atelet restart. + // Absence alone does not prove a pending local checkpoint was detached. + return apierror.FailedPrecondition("atelet on node %s cannot account for local snapshot generation %d", node, knownLocal.GetGeneration()) + } + if err := applyAteletResponseState(ctx, status, state); err != nil { + return err + } + for _, snap := range status.GetSnapshots() { + if snap.GetLocality() == node { + snap.Locality = "" + } + } + for i, report := range []*ateletpb.SnapshotReport{state.GetLatestLocal(), state.GetUploading()} { + if report == nil { + continue + } + reported := report.GetSnapshot() + snap := findSnapshotByGeneration(status.GetSnapshots(), reported.GetGeneration()) + // Suspend can drop an older upload's record while atelet finishes it. + if snap == nil && i == 1 { + continue + } + if snap == nil || snap.GetUuid() != reported.GetSnapshotUuid() || + snap.GetDurableSnapshot().GetObject().GetSnapshotUri() != reported.GetObject().GetSnapshotUri() { + return apierror.FailedPrecondition("atelet on node %s retained an unrecognized snapshot: %v", node, reported) + } + if got := fidelityFromAtelet(reported.GetFidelity()); got == ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_UNSPECIFIED || got != snap.GetFidelity() { + return apierror.FailedPrecondition("atelet on node %s retained snapshot %s with unexpected fidelity", node, snap.GetUuid()) + } + snap.Locality = node + } + return nil +} + +// ensureInProgressSnapshotDiscarded removes uploads of checkpoints that never +// committed locally. A committed local snapshot's upload must continue, even +// while its durable status is still PENDING. func (w *ActorWorkflow) ensureInProgressSnapshotDiscarded(ctx context.Context, actor *ateapi.Actor) (err error) { ctx, done := stepSpan(ctx, "DiscardInProgressSnapshot") defer func() { err = done(err) }() var inProgress []string for _, snap := range actor.GetStatus().GetSnapshots() { + if snap.GetLocality() != "" { + continue + } storage := snap.GetDurableSnapshot() if storage.GetStatus() == ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING && storage.GetObject().GetSnapshotUri() != "" { inProgress = append(inProgress, storage.GetObject().GetSnapshotUri()) @@ -279,14 +335,14 @@ func (w *ActorWorkflow) ensureRevertedFinalized(ctx context.Context, actorRef re storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(latestActor), func(toUpdate *ateapi.Actor) error { toUpdate.Status.State = ateapi.ActorState_ACTOR_STATE_SUSPENDED toUpdate.Status.WorkerAssignment = nil - toUpdate.Status.AssignedNode = "" - // We revert back to the latest durable snapshot, so all local snapshots - // and pending durable snapshots are removed. - // We already cleaned up the node before this in ensureWorkerDiscarded, - // and any in-progress durable snapshot in ensureInProgressSnapshotDiscarded. - // Even during revert, we do not decrease LastAssignedGeneration so that we - // can keep track of stale snapshots that need to be cleaned up. - pruneSnapshots(toUpdate.Status, nil) + local := findLatestLocalSnapshot(toUpdate.Status) + if durable := findLatestDurableSnapshot(toUpdate.Status, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED); durable.GetGeneration() > local.GetGeneration() { + local = nil + } + toUpdate.Status.AssignedNode = local.GetLocality() + // Keep the last successful checkpoint, not a newer incomplete suspend. + // LastAssignedGeneration stays monotonic even when that checkpoint is older. + pruneSnapshots(toUpdate.Status, local) toUpdate.Status.Crash = nil return nil }) diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go index 40d247d296..b3c7dbd2fe 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go @@ -16,6 +16,7 @@ package controlapi import ( "context" + "fmt" "testing" "github.com/google/go-cmp/cmp" @@ -25,6 +26,7 @@ import ( "go.ate.dev/substrate/internal/apierror" "go.ate.dev/substrate/internal/objectstore/objectstoretest" "go.ate.dev/substrate/internal/objectstoreplugin/objectstoreplugintest" + "go.ate.dev/substrate/internal/proto/ateletpb" "go.ate.dev/substrate/internal/resources" "google.golang.org/grpc/codes" "google.golang.org/protobuf/proto" @@ -36,9 +38,7 @@ import ( // // SUSPENDED is the load-bearing rejection: it is revert's own destination, and // also what a suspend that won the race leaves behind, so accepting it would -// report success for the opposite of what was asked. A SUSPENDED actor that -// holds a node-local snapshot is accepted; see -// TestEnsureMarkedReverting_AcceptsPaused. +// report success for the opposite of what was asked. func TestEnsureMarkedReverting_StateMatrix(t *testing.T) { allowed := map[ateapi.ActorState]bool{ ateapi.ActorState_ACTOR_STATE_RUNNING: true, @@ -71,10 +71,8 @@ func TestEnsureMarkedReverting_StateMatrix(t *testing.T) { } } -// TestEnsureMarkedReverting_AcceptsPaused verifies a SUSPENDED actor whose -// latest snapshot is node-local takes the revert edge: the revert discards that -// local copy and returns the actor to its durable snapshot. -func TestEnsureMarkedReverting_AcceptsPaused(t *testing.T) { +// Locality is a cache detail, not an additional lifecycle state. +func TestEnsureMarkedReverting_RejectsPaused(t *testing.T) { ctx := context.Background() st, cleanup := storetest.SetupTestStore(t) defer cleanup() @@ -87,34 +85,26 @@ func TestEnsureMarkedReverting_AcceptsPaused(t *testing.T) { t.Fatalf("get seeded actor: %v", err) } - updated, err := w.ensureMarkedReverting(ctx, actorRef, actor) - if err != nil { - t.Fatalf("ensureMarkedReverting: %v", err) - } - if got := updated.GetStatus().GetState(); got != ateapi.ActorState_ACTOR_STATE_REVERTING { - t.Errorf("state = %v, want REVERTING", got) + if _, err := w.ensureMarkedReverting(ctx, actorRef, actor); apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("ensureMarkedReverting = %v, want FailedPrecondition", err) } } // TestRevertActor_ReturnsActorToItsSnapshot covers the whole workflow from each -// accepted origin: the actor lands SUSPENDED holding the same external snapshot -// it started with, and every pointer to the discarded execution is gone. -// -// None of these actors has an assigned node, so the terminate path is not -// exercised here; it needs the atelet rig, and the unit workflow is built with -// a nil dialer on purpose so an unexpected dial fails loudly. +// accepted origin. A successful local checkpoint survives an interrupted newer +// checkpoint, whether or not an older durable fallback exists. func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { for _, tc := range []struct { - name string - seedState ateapi.ActorState - // paused seeds the local-snapshot shape: the latest snapshot is node-local - // and the actor is pinned to its node. Only meaningful with - // SUSPENDED, which the revert otherwise rejects. - paused bool + name string + seedState ateapi.ActorState + noDurable bool + unrecordedLocal bool }{ {name: "RUNNING", seedState: ateapi.ActorState_ACTOR_STATE_RUNNING}, - {name: "paused", seedState: ateapi.ActorState_ACTOR_STATE_SUSPENDED, paused: true}, + {name: "RUNNING without durable fallback", seedState: ateapi.ActorState_ACTOR_STATE_RUNNING, noDurable: true}, {name: "CRASHED", seedState: ateapi.ActorState_ACTOR_STATE_CRASHED}, + {name: "CRASHED after committing checkpoint", seedState: ateapi.ActorState_ACTOR_STATE_CRASHED, unrecordedLocal: true}, + {name: "CRASHED after first checkpoint", seedState: ateapi.ActorState_ACTOR_STATE_CRASHED, unrecordedLocal: true, noDurable: true}, // A revert that died partway is retried, not rejected. {name: "REVERTING", seedState: ateapi.ActorState_ACTOR_STATE_REVERTING}, } { @@ -124,6 +114,10 @@ func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { st, cleanup := storetest.SetupTestStore(t) defer cleanup() w := newTestActorWorkflow(t, st, "ns", "tmpl1") + wire, atelet := newWireCaptureWorkflow(t, st) + w.dialer = wire.dialer + objects := objectstoretest.New() + w.objectStorePlugin = objectstoreplugintest.ControlClient(objects) actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", seedState) @@ -132,28 +126,41 @@ func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { t.Fatalf("get seeded actor: %v", err) } - // The snapshot revert must preserve, plus the node-local state a - // suspend left behind, which it must not. The suspended actor's - // latest snapshot is the completed local one; the other origins - // also carry the half-written local snapshot an interrupted suspend - // leaves. + // Locality is only committed after a successful checkpoint. The newer + // PENDING snapshot without locality is an interrupted suspend. const keptURI = "gs://snapshots/atespaces/team-a/actors/keep/snapshots/snapshot" tmpl := &ateapi.ActorTemplate{SnapshotConfig: &ateapi.SnapshotConfig{StorageLocation: "gs://snapshots"}} - mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapi.ActorStatus) { + actor = mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapi.ActorStatus) { local1 := newPausedSnapshot(t, 2, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "local-1", "node-1") local1.DurableSnapshot.Object.SnapshotUri = mustActorSnapshotURI(t, tmpl, actor, "local-1").String() inProgress := newPausedSnapshot(t, 3, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "local-in-progress", "") inProgress.DurableSnapshot.Object.SnapshotUri = mustActorSnapshotURI(t, tmpl, actor, "local-in-progress").String() + objects.PutSnapshot(t, mustActorSnapshotURI(t, tmpl, actor, "local-1"), "manifest.json") + objects.PutSnapshot(t, mustActorSnapshotURI(t, tmpl, actor, "local-in-progress"), "manifest.json") + s.AssignedNode = "node-1" + s.WorkerAssignment = wireTestAssignment() s.LastAssignedGeneration = 3 s.Snapshots = []*ateapi.Snapshot{ newDurableSnapshot(1, ateapi.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "kept", keptURI, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED), local1, inProgress, } - if tc.paused { - s.Snapshots[2].Locality = "node-1" + if tc.noDurable { + s.Snapshots = s.Snapshots[1:] + } + if tc.unrecordedLocal { + local1.Locality = "" + if !tc.noDurable { + s.Snapshots[0].Locality = "node-1" + } } }) + seedWireWorker(t, ctx, st, assignmentOf(actor)) + local := findSnapshotByGeneration(actor.GetStatus().GetSnapshots(), 2) + atelet.terminateState = &ateletpb.ActorSnapshotState{LatestLocal: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{ + SnapshotUuid: local.GetUuid(), Generation: local.GetGeneration(), Fidelity: fidelityToAtelet(local.GetFidelity()), + Object: &ateletpb.ObjectStorage{SnapshotUri: local.GetDurableSnapshot().GetObject().GetSnapshotUri()}, + }}} reverted, err := w.RevertActor(ctx, actorRef) if err != nil { @@ -171,26 +178,36 @@ func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { t.Errorf("last_assigned_generation = %d, want 3", got) } gotSnap := findLatestDurableSnapshot(gotStatus, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED) - if got := gotSnap.GetGeneration(); got != 1 { + if got := gotSnap.GetGeneration(); !tc.noDurable && got != 1 { t.Errorf("retained snapshot generation = %d, want 1", got) } - if got := gotSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(); got != keptURI { + if got := gotSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(); !tc.noDurable && got != keptURI { t.Errorf("external snapshot = %q, want it untouched at %q", got, keptURI) } - // Resume prefers a local snapshot over the external one, so a - // surviving LocalSnapshot would restore the execution this - // revert just discarded. - if got := findLatestLocalSnapshot(gotStatus); got != nil { - t.Errorf("local snapshot info = %v, want nil", got) + if got := findLatestLocalSnapshot(gotStatus); got.GetGeneration() != 2 || got.GetLocality() != "node-1" { + t.Errorf("local snapshot = %v, want generation 2 on node-1", got) } - if got := findLatestDurableSnapshot(gotStatus, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING); got != nil { - t.Errorf("in-progress durable snapshot = %v, want nil", got) + if got := findLatestDurableSnapshot(gotStatus, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING); got.GetGeneration() != 2 { + t.Errorf("pending snapshot = %v, want the committed local generation 2", got) } - if got := len(gotStatus.GetSnapshots()); got != 1 { - t.Errorf("len(snapshots) = %d, want 1 (only the completed durable snapshot should remain)", got) + wantSnapshots := 2 + if tc.noDurable { + wantSnapshots = 1 } - if got := gotStatus.GetAssignedNode(); got != "" { - t.Errorf("assigned node = %q, want empty", got) + if got := len(gotStatus.GetSnapshots()); got != wantSnapshots { + t.Errorf("len(snapshots) = %d, want %d", got, wantSnapshots) + } + if got := gotStatus.GetAssignedNode(); got != "node-1" { + t.Errorf("assigned node = %q, want node-1", got) + } + if len(objects.Snapshot(t, mustActorSnapshotURI(t, tmpl, actor, "local-1"))) == 0 { + t.Error("committed local snapshot's pending objects were deleted") + } + if len(objects.Snapshot(t, mustActorSnapshotURI(t, tmpl, actor, "local-in-progress"))) != 0 { + t.Error("incomplete checkpoint's objects survived revert") + } + if !atelet.terminateRequest().GetPreserveSnapshots() { + t.Error("TerminateRequest did not preserve snapshots") } if got := gotStatus.GetWorkerAssignment(); got != nil { t.Errorf("worker assignment = %v, want nil", got) @@ -203,8 +220,7 @@ func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { s.AssignedNode = "node-1" s.WorkerAssignment = wireTestAssignment() }) - seedWireWorker(t, ctx, st, assignmentOf(running)) - wire, atelet := newWireCaptureWorkflow(t, st) + seedAssignment(t, st, wireTestAssignment().GetWorker().GetName(), assignmentOf(running)) suspended, err := wire.SuspendActor(ctx, actorRef, suspendSync) if err != nil { t.Fatalf("SuspendActor after revert: %v", err) @@ -223,6 +239,95 @@ func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { } } +func TestApplyRevertedSnapshotState(t *testing.T) { + for _, tc := range []struct { + name string + change func(*ateletpb.ActorSnapshotState) + wantErr bool + completed bool + registeredUpload bool + }{ + {name: "matching local"}, + {name: "older upload still recorded", registeredUpload: true, change: func(s *ateletpb.ActorSnapshotState) { s.Uploading = ateletReported(t, 1) }}, + {name: "empty tracker with durable checkpoint", completed: true, change: func(s *ateletpb.ActorSnapshotState) { s.LatestLocal = nil }}, + {name: "older upload no longer recorded", change: func(s *ateletpb.ActorSnapshotState) { s.Uploading = ateletReported(t, 1) }}, + {name: "empty tracker", change: func(s *ateletpb.ActorSnapshotState) { s.LatestLocal = nil }, wantErr: true}, + {name: "stale tracker", change: func(s *ateletpb.ActorSnapshotState) { s.LatestLocal = ateletReported(t, 1) }, wantErr: true}, + {name: "different UUID", change: func(s *ateletpb.ActorSnapshotState) { s.LatestLocal.Snapshot.SnapshotUuid = "another" }, wantErr: true}, + {name: "different URI", change: func(s *ateletpb.ActorSnapshotState) { + s.LatestLocal.Snapshot.Object.SnapshotUri = "gs://another/snapshot" + }, wantErr: true}, + {name: "different fidelity", change: func(s *ateletpb.ActorSnapshotState) { + s.LatestLocal.Snapshot.Fidelity = ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES + }, wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + status := &ateapi.ActorStatus{Snapshots: []*ateapi.Snapshot{ + withDurable(t, 2, reportNode, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), + }} + if tc.completed { + status.Snapshots[0].DurableSnapshot.Status = ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED + } + if tc.registeredUpload { + status.Snapshots = append(status.Snapshots, withDurable(t, 1, reportNode, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING)) + } + state := &ateletpb.ActorSnapshotState{LatestLocal: ateletReported(t, 2)} + if tc.change != nil { + tc.change(state) + } + err := applyRevertedSnapshotState(t.Context(), status, reportNode, state) + if tc.wantErr { + if apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("applyRevertedSnapshotState = %v, want FailedPrecondition", err) + } + } else if err != nil { + t.Fatal(err) + } else if got := findLatestLocalSnapshot(status); !tc.completed && got.GetGeneration() != 2 { + t.Errorf("latest local snapshot = %v, want generation 2", got) + } + }) + } +} + +func TestRevertActor_AteletAlreadyDetached(t *testing.T) { + for _, retry := range []bool{false, true} { + t.Run(fmt.Sprintf("retry=%v", retry), func(t *testing.T) { + ctx := t.Context() + st, cleanup := storetest.SetupTestStore(t) + t.Cleanup(cleanup) + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + wire, atelet := newWireCaptureWorkflow(t, st) + w.dialer = wire.dialer + ref := resources.ActorRef{Atespace: "team-a", Name: "id1"} + state := ateapi.ActorState_ACTOR_STATE_CRASHED + if retry { + state = ateapi.ActorState_ACTOR_STATE_REVERTING + } + seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", state, withPausedShape(t, "node-1")) + actor, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatal(err) + } + local := findLatestLocalSnapshot(actor.GetStatus()) + atelet.terminateState = &ateletpb.ActorSnapshotState{LatestDurable: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{ + SnapshotUuid: local.GetUuid(), Generation: local.GetGeneration(), Fidelity: fidelityToAtelet(local.GetFidelity()), + Object: &ateletpb.ObjectStorage{SnapshotUri: local.GetDurableSnapshot().GetObject().GetSnapshotUri()}, + }}} + reverted, err := w.RevertActor(ctx, ref) + if err != nil { + t.Fatal(err) + } + got := reverted.GetStatus() + if got.GetAssignedNode() != "" || findLatestLocalSnapshot(got) != nil { + t.Errorf("stale locality survived completed detach: %v", got) + } + if snap := findLatestDurableSnapshot(got, ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED); snap.GetUuid() != local.GetUuid() { + t.Errorf("completed upload lost: got %v, want %v", snap, local) + } + }) + } +} + func TestRevertActor_UnreachableNode(t *testing.T) { for _, tt := range []struct { name string @@ -251,7 +356,7 @@ func TestRevertActor_UnreachableNode(t *testing.T) { wire, atelet := newWireCaptureWorkflow(t, st) w.dialer = wire.dialer ref := resources.ActorRef{Atespace: "team-a", Name: "id1"} - seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", ateapi.ActorState_ACTOR_STATE_SUSPENDED) + seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", ateapi.ActorState_ACTOR_STATE_CRASHED) actor, err := st.GetActor(ctx, ref) if err != nil { t.Fatal(err) @@ -336,23 +441,21 @@ func TestRevertActor_UnreachableNode(t *testing.T) { } } -// TestRevertActor_PausedSnapshotAlreadyUploaded verifies a revert of a -// suspended actor whose local snapshot atelet uploaded returns it to that -// upload, its latest durable snapshot, and collects the durable snapshot the -// upload replaced. The actor has no assigned node, so there is nothing to -// terminate, as in TestRevertActor_ReturnsActorToItsSnapshot. -func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) { +// A completed upload does not turn its local copy into disposable execution state. +func TestRevertActor_LocalSnapshotAlreadyUploaded(t *testing.T) { memory := ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY completed := ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED ctx := context.Background() st, cleanup := storetest.SetupTestStore(t) defer cleanup() w := newTestActorWorkflow(t, st, "ns", "tmpl1") + wire, atelet := newWireCaptureWorkflow(t, st) + w.dialer = wire.dialer objects := objectstoretest.New() w.objectStorePlugin = objectstoreplugintest.ControlClient(objects) actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} - seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapi.ActorState_ACTOR_STATE_SUSPENDED) + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapi.ActorState_ACTOR_STATE_CRASHED) actor, err := st.GetActor(ctx, actorRef) if err != nil { t.Fatalf("get seeded actor: %v", err) @@ -363,9 +466,14 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) { uploaded := mustActorSnapshotURI(t, tmpl, actor, "pause-snap") objects.PutSnapshot(t, old, "manifest.json") objects.PutSnapshot(t, uploaded, "manifest.json") + atelet.terminateState = &ateletpb.ActorSnapshotState{LatestLocal: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{ + SnapshotUuid: "pause-snap", Generation: 2, Fidelity: fidelityToAtelet(memory), + Object: &ateletpb.ObjectStorage{SnapshotUri: uploaded.String()}, + }}} mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapi.ActorStatus) { paused := newPausedSnapshot(t, 2, memory, "", "pause-snap", "node-1") paused.DurableSnapshot = &ateapi.SnapshotStorage{Status: completed, Object: &ateapi.ObjectSnapshot{SnapshotUri: uploaded.String()}} + s.AssignedNode = "node-1" s.LastAssignedGeneration = 2 s.Snapshots = []*ateapi.Snapshot{ newDurableSnapshot(1, ateapi.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, memory, "", old.Name(), old.String(), completed), @@ -377,7 +485,7 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) { if err != nil { t.Fatalf("RevertActor: %v", err) } - want := []snapshotOutcome{{generation: 2, fidelity: memory, durable: completed}} + want := []snapshotOutcome{{generation: 2, fidelity: memory, durable: completed, locality: "node-1"}} if diff := cmp.Diff(want, describeSnapshots(reverted.GetStatus()), cmp.AllowUnexported(snapshotOutcome{})); diff != "" { t.Errorf("snapshots mismatch (-want +got):\n%s", diff) } @@ -387,6 +495,9 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) { if len(objects.Snapshot(t, uploaded)) == 0 { t.Error("the upload the revert returned the actor to was released") } + if req := atelet.terminateRequest(); !req.GetPreserveSnapshots() || req.GetWorkerPodUid() != "" { + t.Errorf("TerminateRequest = %v, want snapshot preservation without a worker", req) + } } // TestRevertActor_RejectsSuspended covers both ways an actor can already be @@ -394,6 +505,45 @@ func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) { // finished. Both are FailedPrecondition — revert is not a no-op success on its // own destination, for the same reason a second DeleteActor is NotFound. func TestRevertActor_RejectsSuspended(t *testing.T) { + for _, tc := range []struct { + name string + locality string + status ateapi.SnapshotStorageStatus + }{ + {name: "local pending", locality: "node-1", status: ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING}, + {name: "local uploaded", locality: "node-1", status: ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED}, + {name: "detached uploaded", status: ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := t.Context() + st, cleanup := storetest.SetupTestStore(t) + t.Cleanup(cleanup) + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + wire, _ := newWireCaptureWorkflow(t, st) + w.dialer = wire.dialer + ref := resources.ActorRef{Atespace: "team-a", Name: "id1"} + seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", ateapi.ActorState_ACTOR_STATE_SUSPENDED, withPausedShape(t, "node-1")) + actor, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatal(err) + } + actor = mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapi.ActorStatus) { + s.AssignedNode = tc.locality + s.Snapshots[0].Locality = tc.locality + s.Snapshots[0].DurableSnapshot.Status = tc.status + }) + if _, err := w.RevertActor(ctx, ref); apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("RevertActor = %v, want FailedPrecondition", err) + } + stored, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatal(err) + } + if !proto.Equal(stored, actor) { + t.Errorf("rejected revert changed actor: got %v, want %v", stored, actor) + } + }) + } t.Run("actor was already suspended", func(t *testing.T) { ctx := context.Background() st, cleanup := storetest.SetupTestStore(t) @@ -499,6 +649,7 @@ func TestEnsureInProgressSnapshotDiscarded(t *testing.T) { // earlierPause records an older PENDING snapshot: an async suspend whose // upload atelet never reported before the actor resumed. earlierPause bool + localEarlierPause bool wantInFlightDiscarded bool wantEarlierPauseDiscarded bool wantErr bool @@ -524,6 +675,13 @@ func TestEnsureInProgressSnapshotDiscarded(t *testing.T) { foreignInFlight: true, wantErr: true, }, + { + name: "retains a committed local snapshot while discarding an incomplete checkpoint", + inFlight: inFlightSnapshotName, + earlierPause: true, + localEarlierPause: true, + wantInFlightDiscarded: true, + }, { name: "discards an earlier pause's unrecorded upload along with the in-flight snapshot", inFlight: inFlightSnapshotName, @@ -586,6 +744,10 @@ func TestEnsureInProgressSnapshotDiscarded(t *testing.T) { if tt.earlierPause { s.LastAssignedGeneration++ s.Snapshots = append(s.Snapshots, newDurableSnapshot(s.LastAssignedGeneration, ateapi.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", earlierPause.Name(), earlierPause.String(), ateapi.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING)) + if tt.localEarlierPause { + s.Snapshots[len(s.Snapshots)-1].Locality = "node-1" + s.AssignedNode = "node-1" + } } if tt.inFlight != "" { s.LastAssignedGeneration++ diff --git a/cmd/atelet/internal/apivalidation/zz_generated.validation.go b/cmd/atelet/internal/apivalidation/zz_generated.validation.go index a312fc66a6..206fa57611 100644 --- a/cmd/atelet/internal/apivalidation/zz_generated.validation.go +++ b/cmd/atelet/internal/apivalidation/zz_generated.validation.go @@ -3790,6 +3790,34 @@ func Validate_TerminateRequest( errs = append(errs, fn(fldPath.Child("spec"), obj.Spec, oldVal, oldObj != nil)...) } + { // field ateletpb.TerminateRequest.PreserveSnapshots + fn := func( + fldPath *field.Path, + obj, oldObj *bool, + oldValueCorrelated bool) (errs field.ErrorList) { + // don't revalidate unchanged data + if oldValueCorrelated && op.Type == operation.Update { + if obj == oldObj || (obj != nil && oldObj != nil && *obj == *oldObj) { + return nil + } + } + // call field-attached validations + earlyReturn := false + if e := validate.OptionalValue(ctx, op, fldPath, obj, oldObj).MarkShortCircuit(); len(e) != 0 { + earlyReturn = true + } + if earlyReturn { + return // do not proceed + } + return + } + oldVal := safe.Field(oldObj, + func(oldObj *ateletpb.TerminateRequest) *bool { + return &oldObj.PreserveSnapshots + }) + errs = append(errs, fn(fldPath.Child("preserve_snapshots"), &obj.PreserveSnapshots, oldVal, oldObj != nil)...) + } + return errs } diff --git a/cmd/atelet/internal/snapshottracker/detach_test.go b/cmd/atelet/internal/snapshottracker/detach_test.go index 5b073f86d5..27b2655cd7 100644 --- a/cmd/atelet/internal/snapshottracker/detach_test.go +++ b/cmd/atelet/internal/snapshottracker/detach_test.go @@ -432,6 +432,74 @@ func TestTrackerDetachTimer(t *testing.T) { } } +// Terminating a resumed actor restores its detach deadline without postponing +// the upload or changing its committed snapshot. +func TestTrackerScheduleDetachAfterTermination(t *testing.T) { + tr := newIdleTestTracker(t, newFakeSnapshotStore()) + want := tr.CommitLocal(trackedActorUID, testRef("s1", 1, tr.clock.Now())) + end, err := tr.BeginStart(t.Context(), trackedActorUID) + if err != nil { + t.Fatal(err) + } + end() + endTerminate := tr.BeginTerminate(trackedActorUID) + defer endTerminate() + if err := tr.ScheduleDetach(trackedActorUID); err != nil { + t.Fatal(err) + } + due := detachDueOf(tr, trackedActorUID) + if expected := tr.clock.Now().Add(tr.detachDelay); !due.Equal(expected) { + t.Fatalf("detach due = %v, want %v", due, expected) + } + if got := detachDueOf(newTracker(t), trackedActorUID); !got.Equal(due) { + t.Errorf("persisted detach due = %v, want %v", got, due) + } + if got := tr.State(trackedActorUID); !proto.Equal(got, want) { + t.Errorf("snapshot state changed: got %v, want %v", got, want) + } + stepTo(tr, tr.clock.Now().Add(time.Minute)) + if err := tr.ScheduleDetach(trackedActorUID); err != nil { + t.Fatal(err) + } + if got := detachDueOf(tr, trackedActorUID); !got.Equal(due) { + t.Errorf("retry postponed detach: got %v, want %v", got, due) + } +} + +func TestTrackerScheduleDetachRetriesFailedSave(t *testing.T) { + tr := newIdleTestTracker(t, newFakeSnapshotStore()) + tr.CommitLocal(trackedActorUID, testRef("s1", 1, tr.clock.Now())) + end, err := tr.BeginStart(t.Context(), trackedActorUID) + if err != nil { + t.Fatal(err) + } + end() + endTerminate := tr.BeginTerminate(trackedActorUID) + defer endTerminate() + stateFile := ateletpath.ActorSnapshotStateFile(trackedActorUID) + if err := os.Remove(stateFile); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(stateFile, 0o700); err != nil { + t.Fatal(err) + } + if err := tr.ScheduleDetach(trackedActorUID); apierror.Code(err) != codes.Unavailable { + t.Fatalf("ScheduleDetach = %v, want Unavailable", err) + } + if due := detachDueOf(tr, trackedActorUID); !due.IsZero() { + t.Fatalf("failed save scheduled detach at %v", due) + } + if err := os.Remove(stateFile); err != nil { + t.Fatal(err) + } + if err := tr.ScheduleDetach(trackedActorUID); err != nil { + t.Fatal(err) + } + if due := detachDueOf(newTracker(t), trackedActorUID); due.IsZero() { + t.Fatal("retry did not persist a detach deadline") + } +} + // A start that cannot save that it began fails before anything starts, and // leaves the detach scheduled and the actor not started. func TestTrackerStartFailsIfItCannotSave(t *testing.T) { diff --git a/cmd/atelet/internal/snapshottracker/lifecycle.go b/cmd/atelet/internal/snapshottracker/lifecycle.go index 8e9284002b..ec1d1d1e27 100644 --- a/cmd/atelet/internal/snapshottracker/lifecycle.go +++ b/cmd/atelet/internal/snapshottracker/lifecycle.go @@ -312,6 +312,26 @@ func (t *Tracker) BeginTerminate(actorUID string) (end func()) { return t.holdLocked(actorUID, t.findLocked(actorUID), holdTerminate) } +// ScheduleDetach makes a stopped actor's local snapshot eligible for detach +// again after BeginStart unscheduled it. It leaves uploads and their original +// deadlines intact. Retrying a termination does not postpone an existing detach. +func (t *Tracker) ScheduleDetach(actorUID string) error { + t.lock(actorUID) + defer t.unlock(actorUID) + a := t.findLocked(actorUID) + if a == nil || a.LatestLocal == nil || !a.DetachAt.IsZero() { + return nil + } + unsaved := a.unsaved + a.DetachAt = t.clock.Now().Add(t.detachDelay) + if err := t.writeStateLocked(a); err != nil { + a.DetachAt, a.unsaved = time.Time{}, unsaved + return apierror.Unavailable("while scheduling the detach of actor %s after stopping it: %v", actorUID, err) + } + t.requeueLocked(a) + return nil +} + // holdLocked records a Run, Restore, Checkpoint, PromoteSnapshot, or Terminate // of the actor in progress on its entry a, which holds detaches and state // reports of the actor off (and background uploads when kind.blocksUploads), diff --git a/cmd/atelet/lifecycle_test.go b/cmd/atelet/lifecycle_test.go index 1d96035c65..5a2092cac4 100644 --- a/cmd/atelet/lifecycle_test.go +++ b/cmd/atelet/lifecycle_test.go @@ -3115,19 +3115,62 @@ func TestImmediateCheckpointHoldsOffBackgroundUpload(t *testing.T) { } } -// TestTerminateHoldsOffBackgroundUpload terminates an actor resumed from its -// local snapshot, whose background upload comes due while the worker tears the -// actor down: the upload does not start, since Terminate would only cancel it. -// When the worker fails to tear the actor down, Terminate fails before it -// forgets the actor, and the upload, still due, starts once Terminate returns. +// Revert needs the retained state even when a previous detach already finished. +func TestTerminateRetainsSnapshotState(t *testing.T) { + for _, tc := range []struct { + name string + mode ateletpb.UploadMode + detached bool + }{ + {name: "pending local", mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED}, + {name: "completed local", mode: ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE}, + {name: "already detached", mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED, detached: true}, + } { + t.Run(tc.name, func(t *testing.T) { + a := runPluginActor(t) + a.s.snapshots = snapshottracker.New(t.Context(), &trackerStore{s: a.s}, clock.RealClock{}) + if _, err := a.checkpoint(t, a.snapshot("revert-snap", 1), tc.mode); err != nil { + t.Fatal(err) + } + workerUID := pluginWorkerPodUID + if tc.detached { + if _, err := a.s.DetachActor(t.Context(), &ateletpb.DetachActorRequest{Atespace: "ate-demo", ActorName: "counter", ActorUid: pluginActorUID}); err != nil { + t.Fatal(err) + } + workerUID = "" + } else { + a.restoreLocal(t, "revert-snap", 1) + } + want := a.s.snapshots.State(pluginActorUID) + resp, err := a.s.Terminate(t.Context(), &ateletpb.TerminateRequest{ + Atespace: "ate-demo", ActorName: "counter", ActorUid: pluginActorUID, + WorkerPodUid: workerUID, Spec: a.spec, PreserveSnapshots: true, + }) + if err != nil { + t.Fatal(err) + } + if !proto.Equal(resp.GetState(), want) { + t.Errorf("retained snapshot state = %v, want %v", resp.GetState(), want) + } + if !tc.detached { + a.restoreLocal(t, "revert-snap", 1) + } + }) + } +} + +// Terminate holds off new uploads. Delete cancels them, while Revert and failed +// teardown let pending uploads continue after the call. func TestTerminateHoldsOffBackgroundUpload(t *testing.T) { for _, tc := range []struct { name string // teardownErr is what the worker's TerminateWorkload fails with. - teardownErr error + teardownErr error + preserveSnapshots bool }{ {name: "Terminate succeeds"}, {name: "Terminate fails", teardownErr: status.Error(codes.Internal, "teardown failed")}, + {name: "Revert preserves pending upload", preserveSnapshots: true}, } { t.Run(tc.name, func(t *testing.T) { a, store, stepPastDue := resumedWithPendingUpload(t) @@ -3145,6 +3188,7 @@ func TestTerminateHoldsOffBackgroundUpload(t *testing.T) { ActorTemplateName: "counter", WorkerPodUid: pluginWorkerPodUID, Spec: a.spec, + PreserveSnapshots: tc.preserveSnapshots, }) switch { case tc.teardownErr == nil && err != nil: @@ -3155,7 +3199,7 @@ func TestTerminateHoldsOffBackgroundUpload(t *testing.T) { if started.Load() { t.Error("background upload of pause-snap-1 started during Terminate, want none") } - if tc.teardownErr == nil { + if tc.teardownErr == nil && !tc.preserveSnapshots { select { case <-store.started: t.Error("background upload of pause-snap-1 started during Terminate, want none") @@ -3163,6 +3207,16 @@ func TestTerminateHoldsOffBackgroundUpload(t *testing.T) { } return } + if tc.preserveSnapshots { + if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-1")); err != nil { + t.Fatalf("committed local checkpoint after Revert termination: %v", err) + } + if _, err := os.Stat(ateletpath.ActorSnapshotStateFile(pluginActorUID)); err != nil { + t.Fatalf("persisted snapshot state after Revert termination: %v", err) + } + // Restoring the same snapshot also checks its manifest and sandbox assets. + a.restoreLocal(t, "pause-snap-1", 1) + } select { case <-store.started: case <-time.After(10 * time.Second): diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 96d5c0789b..8b0b88ec17 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -2019,6 +2019,18 @@ func (s *Atelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) return nil, fmt.Errorf("failed to unmount external volumes during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) } + if req.GetPreserveSnapshots() { + // Workload teardown must not cancel uploads or remove the checkpoint and + // sandbox assets needed to restore the last successful suspend. + if err := resetActorDirs(actorUID); err != nil { + return nil, fmt.Errorf("failed to reset actor directories during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + if err := s.snapshots.ScheduleDetach(actorUID); err != nil { + return nil, err + } + return &ateletpb.TerminateResponse{State: s.snapshots.State(actorUID)}, nil + } + // The actor is gone, so no local snapshot of it can ever be restored again. // Forget cancels the actor's background upload or detach in progress and // waits for it to end before this prunes the actor's files. diff --git a/cmd/kubectl-ate/README.md b/cmd/kubectl-ate/README.md index 21c27a77e4..9ed2f37231 100644 --- a/cmd/kubectl-ate/README.md +++ b/cmd/kubectl-ate/README.md @@ -214,7 +214,7 @@ kubectl ate suspend actor my-actor -a # Suspend an actor and wait for its snapshot to reach storage. kubectl ate suspend actor my-actor -a --mode=sync -# Revert an actor to its last external snapshot (discards live or crashed state, or suspended state with a node-local snapshot, and returns to SUSPENDED) +# Revert a running or crashed actor to its latest committed snapshot, including a local snapshot whose upload is pending # A suspended actor without a node-local snapshot is already at its external snapshot, so it cannot be reverted kubectl ate revert actor my-actor -a diff --git a/cmd/kubectl-ate/internal/cmd/revert_actor.go b/cmd/kubectl-ate/internal/cmd/revert_actor.go index 09d06219d0..28057191a4 100644 --- a/cmd/kubectl-ate/internal/cmd/revert_actor.go +++ b/cmd/kubectl-ate/internal/cmd/revert_actor.go @@ -30,10 +30,9 @@ var revertActorCmd = &cobra.Command{ Short: "Revert an actor to its latest snapshot", Long: `Revert an actor to SUSPENDED, discarding its current execution. -The actor's external snapshot is left untouched, so a later resume restores -from it. Only crashed or running actors, or suspended actors with a node-local -snapshot, can be reverted. A suspended actor without a node-local snapshot is -already at its durable snapshot, so there is nothing to revert.`, +The latest committed snapshot is kept for the next resume, including a local +snapshot whose upload is pending. Only crashed or running actors can be reverted. +An already suspended actor cannot be reverted, regardless of snapshot locality.`, Args: exactlyOneArg("", "atespace"), RunE: func(cmd *cobra.Command, args []string) error { ctx := cmd.Context() diff --git a/docs/api-guide.md b/docs/api-guide.md index ef7a3db7af..b8ad19941e 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -624,7 +624,7 @@ Every actor that must keep its state through an eviction needs a `SIGTERM` handl - **runs in the container's main process.** Substrate signals only to the main process. - **gets the actor suspended within the 30 minutes.** -An actor still running when the 30 minutes are up is killed. Once the pod is gone, the actor moves to `ACTOR_STATE_CRASHED`, and everything since its last snapshot is lost. `RevertActor` (`kubectl ate revert`) returns a crashed actor to `ACTOR_STATE_SUSPENDED` at its last external snapshot. +An actor still running when the 30 minutes are up is killed. Once the pod is gone, the actor moves to `ACTOR_STATE_CRASHED`, and everything since its last snapshot is lost. `RevertActor` (`kubectl ate revert`) returns a crashed actor to `ACTOR_STATE_SUSPENDED` at its last committed snapshot, including a local snapshot whose upload is pending. The node can cut the 30 minutes short. GKE node upgrades and cluster autoscaler scale-downs wait up to an hour for a pod, so they leave the full 30 minutes. A reclaimed GKE [Spot VM](https://docs.cloud.google.com/kubernetes-engine/docs/concepts/spot-vms) gives pods 15 seconds by default and at most 2 minutes, so an actor there has to be suspended in that time or it crashes. @@ -697,11 +697,11 @@ Deletion always runs before the database reference is dropped, and a failure fai > **Do not delete a tag while actors created from it exist.** A clone borrows the tag's snapshot rather than copying it, and only stops borrowing at its own first suspend (its `status.snapshots` entry still names the tag's prefix while it is). Deleting the tag leaves such a clone unable to resume. This is not prevented today. #### `RevertActor` -Discards an actor's live or crashed execution and transitions it to `ACTOR_STATE_SUSPENDED` at its last completed external snapshot (`status.snapshots`). +Discards an actor's live or crashed execution and transitions it to `ACTOR_STATE_SUSPENDED` at its last committed snapshot (`status.snapshots`). * **Request:** `RevertActorRequest` - * `actor`: `ObjectRef` of the actor to revert. Accepted from `ACTOR_STATE_RUNNING` and `ACTOR_STATE_CRASHED` (plus `ACTOR_STATE_REVERTING` for idempotent retries), and from `ACTOR_STATE_SUSPENDED` when the actor's latest snapshot is node-local. Calling `RevertActor` on an `ACTOR_STATE_SUSPENDED` actor with no node-local snapshot returns `FAILED_PRECONDITION`. + * `actor`: `ObjectRef` of the actor to revert. Accepted from `ACTOR_STATE_RUNNING` and `ACTOR_STATE_CRASHED` (plus `ACTOR_STATE_REVERTING` for idempotent retries). Calling `RevertActor` on an `ACTOR_STATE_SUSPENDED` actor returns `FAILED_PRECONDITION`, regardless of whether its snapshot is local or uploaded. * **Response:** `RevertActorResponse` containing the reverted `Actor` in `ACTOR_STATE_SUSPENDED`. -* Reverting terminates any bound worker sandbox, clears node-local checkpoints, and garbage-collects any partial external snapshot left by an interrupted suspend while preserving the last committed durable snapshot in `status.snapshots`. +* Reverting terminates any bound worker sandbox and releases its assignment. It keeps the latest committed local snapshot and its background upload, as well as the latest completed external snapshot. Partial external snapshots from interrupted suspends that never committed locally are garbage-collected. A retained local snapshot keeps the actor pinned to that node until it is detached; snapshots on a deleted node cannot be retained. * External volumes are not reverted. Their contents are never part of a snapshot, so a reverted actor comes back with its memory and root filesystem rewound but its volumes exactly as the discarded execution left them. #### `DeleteActor` diff --git a/docs/architecture.md b/docs/architecture.md index 708aaea427..69815fcd32 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -512,7 +512,7 @@ tags cannot be deleted until they are. ### Phase 4: Recovery (`RevertActor`) and Deletion (`DeleteActor`) -An actor in `ACTOR_STATE_RUNNING` or `ACTOR_STATE_CRASHED`, or an `ACTOR_STATE_SUSPENDED` actor with a node-local latest snapshot, can be reverted back to `ACTOR_STATE_SUSPENDED` at its durable snapshot via `RevertActor`. Reverting terminates any active sandbox, discards any node-local checkpoint or partial in-progress snapshot, and preserves the actor's last completed external snapshot so a subsequent `ResumeActor` restores from that checkpoint. +An actor in `ACTOR_STATE_RUNNING` or `ACTOR_STATE_CRASHED` can be reverted back to `ACTOR_STATE_SUSPENDED` via `RevertActor`. Reverting terminates any active sandbox while retaining the latest committed local snapshot and its background upload, plus the latest completed external snapshot. It cleans up partial snapshots from interrupted suspends that never committed locally. A subsequent `ResumeActor` restores the retained checkpoint. Already suspended actors cannot be reverted, regardless of snapshot locality. An actor in any state (such as `ACTOR_STATE_RUNNING`) can be deleted from the Control Plane. The workflow terminates the running containers on the worker, detaches mounted volumes, and frees the worker assignment before deleting the record. diff --git a/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go b/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go index 3c37ea65e5..8b07036600 100644 --- a/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go +++ b/internal/e2e/suites/combinedvolumes/combinedvolumes_test.go @@ -461,7 +461,7 @@ func TestCombinedVolumes(t *testing.T) { }) }) - // A revert rewinds to the last external snapshot. + // A revert rewinds to the last committed snapshot, retaining its local copy. t.Run("RevertRewindsOnlySnapshottedVolumes", func(t *testing.T) { // Written after the suspend above, so it is in no snapshot the revert // can return the actor to. @@ -477,7 +477,13 @@ func TestCombinedVolumes(t *testing.T) { if got := reverted.GetActor().GetStatus().GetState(); got != ateapi.ActorState_ACTOR_STATE_SUSPENDED { t.Fatalf("state after revert = %v, want SUSPENDED", got) } - requireAssignedNode(t, reverted.GetActor(), "") + localNode := "" + for _, snap := range reverted.GetActor().GetStatus().GetSnapshots() { + if snap.GetLocality() != "" { + localNode = snap.GetLocality() + } + } + requireAssignedNode(t, reverted.GetActor(), localNode) resumeCtx, cancel := context.WithTimeout(ctx, 2*time.Minute) defer cancel() diff --git a/internal/e2e/suites/demo/demo_test.go b/internal/e2e/suites/demo/demo_test.go index 166b77a33c..08f47e791b 100644 --- a/internal/e2e/suites/demo/demo_test.go +++ b/internal/e2e/suites/demo/demo_test.go @@ -1169,46 +1169,50 @@ func revertActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj } validateCounterResponse(t, resp, "after revert from RUNNING", 2, 2) - // Revert from async suspend (local snapshot). Resume prefers a local - // snapshot over the external one, so clearing the local snapshot is what - // prevents this async suspend's count=2 checkpoint from restoring on the - // next resume. - if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: actorRef, Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_ASYNCHRONOUS}}); err != nil { + // A newer async checkpoint must survive revert, even before its upload. + paused, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: actorRef, Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_ASYNCHRONOUS}}) + if err != nil { t.Fatalf("failed to pause Actor: %v", err) } waitForActorPaused(ctx, t, clients, actorName) - - reverted, err = clients.SubstrateAPI.RevertActor(ctx, &ateapi.RevertActorRequest{Actor: actorRef}) - if err != nil { - t.Fatalf("RevertActor from paused: %v", err) + checkpointGeneration := paused.GetActor().GetStatus().GetLastAssignedGeneration() + // SUSPENDED is rejected even while the snapshot is still local. + if _, err := clients.SubstrateAPI.RevertActor(ctx, &ateapi.RevertActorRequest{Actor: actorRef}); status.Code(err) != codes.FailedPrecondition { + t.Fatalf("RevertActor from SUSPENDED: got %v, want FailedPrecondition", err) } - if got := reverted.GetActor().GetStatus().GetState(); got != ateapi.ActorState_ACTOR_STATE_SUSPENDED { - t.Fatalf("state after revert from paused = %v, want SUSPENDED", got) + if _, err := e2e.ResumeActorAwaitCapacity(t, ctx, clients, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { + t.Fatalf("failed to resume Actor after async suspend: %v", err) + } + waitForActorState(ctx, t, clients, actorName, ateapi.ActorState_ACTOR_STATE_RUNNING) + resp, err = callActor(t, resources.ActorRef{Atespace: demoAtespace, Name: actorName}) + if err != nil { + t.Fatalf("failed to call actor before async revert: %v", err) } - if hasLocalSnapshot(reverted.GetActor().GetStatus()) { - t.Fatal("local snapshot pointer survived revert from paused") + validateCounterResponse(t, resp, "before revert from async snapshot", 3, 3) + reverted, err = clients.SubstrateAPI.RevertActor(ctx, &ateapi.RevertActorRequest{Actor: actorRef}) + if err != nil { + t.Fatalf("RevertActor with async snapshot: %v", err) } - if got := durableSnapshotURI(reverted.GetActor().GetStatus()); got != snapshotURI { - t.Fatalf("snapshot URI after revert from paused = %q, want %q", got, snapshotURI) + var kept bool + for _, snap := range reverted.GetActor().GetStatus().GetSnapshots() { + if snap.GetGeneration() == checkpointGeneration { + kept = true + } } - - // SUSPENDED is rejected with FailedPrecondition rather than no-oping. - if _, err := clients.SubstrateAPI.RevertActor(ctx, &ateapi.RevertActorRequest{Actor: actorRef}); status.Code(err) != codes.FailedPrecondition { - t.Fatalf("RevertActor from SUSPENDED: got %v, want FailedPrecondition", err) + if !kept { + t.Fatalf("async checkpoint generation %d lost after revert: %v", checkpointGeneration, reverted.GetActor().GetStatus()) } - - // Final resume: if the local checkpoint had been restored, the counter - // would advance from 2 to 3. Expecting 2 proves the external snapshot won. if _, err := e2e.ResumeActorAwaitCapacity(t, ctx, clients, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { - t.Fatalf("failed to resume Actor after paused revert: %v", err) + t.Fatalf("failed to resume Actor after async revert: %v", err) } waitForActorState(ctx, t, clients, actorName, ateapi.ActorState_ACTOR_STATE_RUNNING) resp, err = callActor(t, resources.ActorRef{Atespace: demoAtespace, Name: actorName}) if err != nil { - t.Fatalf("failed to call actor after paused revert: %v", err) + t.Fatalf("failed to call actor after async revert: %v", err) } - validateCounterResponse(t, resp, "after revert from paused", 2, 2) + // 2 means the older durable snapshot won; 4 means the live execution survived. + validateCounterResponse(t, resp, "after revert from async snapshot", 3, 3) if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapi.DeleteActorRequest{Actor: actorRef}); err != nil { t.Fatalf("failed to delete Actor: %v", err) @@ -1839,8 +1843,8 @@ func TestRevertCrashedActor(t *testing.T) { if reverted.GetActor().GetStatus().GetWorkerAssignment() != nil { t.Fatal("worker assignment survived revert") } - if got := reverted.GetActor().GetStatus().GetAssignedNode(); got != "" { - t.Fatalf("assigned node after revert = %q, want empty", got) + if got, want := reverted.GetActor().GetStatus().GetAssignedNode(), localSnapshot(crashed.GetStatus()).GetLocality(); got != want { + t.Fatalf("assigned node after revert = %q, want %q", got, want) } if _, err := e2e.ResumeActorAwaitCapacity(t, ctx, clients, &ateapi.ResumeActorRequest{Actor: actorRef}); err != nil { diff --git a/internal/proto/ateletpb/atelet.proto b/internal/proto/ateletpb/atelet.proto index f68551c7b6..0ffe7b6375 100644 --- a/internal/proto/ateletpb/atelet.proto +++ b/internal/proto/ateletpb/atelet.proto @@ -519,9 +519,19 @@ message TerminateRequest { // // +k8s:optional WorkloadSpec spec = 7; + + // Keep committed local snapshots and their background uploads after stopping + // the workload. Revert uses these snapshots for the next resume. + // + // +k8s:optional + bool preserve_snapshots = 8; } message TerminateResponse { + // Snapshot state retained when preserve_snapshots is set. + // + // +k8s:optional + ActorSnapshotState state = 1; } message RunRequest { From 9ac133aed7f827a5074366ed69dbf85ab8f73efb Mon Sep 17 00:00:00 2001 From: cloudonly Date: Wed, 21 Oct 2026 15:28:23 +0800 Subject: [PATCH 2/4] Stage local checkpoints until their manifests are complete --- api/ateapi/v1/ateapi.pb.go | 4 +- api/ateapi/v1/ateapi.proto | 4 +- cmd/atelet/lifecycle_test.go | 65 ++++++++- cmd/atelet/main.go | 2 +- cmd/kubectl-ate/README.md | 2 +- internal/proto/ateletpb/atelet.pb.go | 196 +++++++++++++++------------ 6 files changed, 179 insertions(+), 94 deletions(-) diff --git a/api/ateapi/v1/ateapi.pb.go b/api/ateapi/v1/ateapi.pb.go index 250a2f5968..78234c4987 100644 --- a/api/ateapi/v1/ateapi.pb.go +++ b/api/ateapi/v1/ateapi.pb.go @@ -5383,8 +5383,8 @@ func (x *RevertActorRequest) GetActor() *ObjectRef { // RevertActorResponse holds the reverted Actor. type RevertActorResponse struct { state protoimpl.MessageState `protogen:"open.v1"` - // actor as it stands after the revert, SUSPENDED and still holding the - // external snapshot it was reverted to. + // Actor after the revert, SUSPENDED and retaining its latest committed + // snapshot, which may be local with a pending upload or already durable. Actor *Actor `protobuf:"bytes,1,opt,name=actor,proto3" json:"actor,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/api/ateapi/v1/ateapi.proto b/api/ateapi/v1/ateapi.proto index d84302fec7..89d07ef893 100644 --- a/api/ateapi/v1/ateapi.proto +++ b/api/ateapi/v1/ateapi.proto @@ -1865,8 +1865,8 @@ message RevertActorRequest { // RevertActorResponse holds the reverted Actor. message RevertActorResponse { - // actor as it stands after the revert, SUSPENDED and still holding the - // external snapshot it was reverted to. + // Actor after the revert, SUSPENDED and retaining its latest committed + // snapshot, which may be local with a pending upload or already durable. Actor actor = 1; } diff --git a/cmd/atelet/lifecycle_test.go b/cmd/atelet/lifecycle_test.go index 5a2092cac4..cb79f0b17b 100644 --- a/cmd/atelet/lifecycle_test.go +++ b/cmd/atelet/lifecycle_test.go @@ -3118,13 +3118,15 @@ func TestImmediateCheckpointHoldsOffBackgroundUpload(t *testing.T) { // Revert needs the retained state even when a previous detach already finished. func TestTerminateRetainsSnapshotState(t *testing.T) { for _, tc := range []struct { - name string - mode ateletpb.UploadMode - detached bool + name string + mode ateletpb.UploadMode + detached bool + lostTracker bool }{ {name: "pending local", mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED}, {name: "completed local", mode: ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE}, {name: "already detached", mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED, detached: true}, + {name: "tracker state lost", mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED, lostTracker: true}, } { t.Run(tc.name, func(t *testing.T) { a := runPluginActor(t) @@ -3141,6 +3143,9 @@ func TestTerminateRetainsSnapshotState(t *testing.T) { } else { a.restoreLocal(t, "revert-snap", 1) } + if tc.lostTracker { + a.s.snapshots = snapshottracker.New(t.Context(), &trackerStore{s: a.s}, clock.RealClock{}) + } want := a.s.snapshots.State(pluginActorUID) resp, err := a.s.Terminate(t.Context(), &ateletpb.TerminateRequest{ Atespace: "ate-demo", ActorName: "counter", ActorUid: pluginActorUID, @@ -3159,6 +3164,60 @@ func TestTerminateRetainsSnapshotState(t *testing.T) { } } +func TestTerminateDiscardsIncompleteLocalCheckpoint(t *testing.T) { + for _, keepCommitted := range []bool{false, true} { + t.Run(fmt.Sprintf("committed=%t", keepCommitted), func(t *testing.T) { + a := runPluginActor(t) + if keepCommitted { + if _, err := a.checkpoint(t, a.snapshot("committed-snap", 1), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil { + t.Fatal(err) + } + a.restoreLocal(t, "committed-snap", 1) + } + want := a.s.snapshots.State(pluginActorUID) + // Exercise the local writer directly to force a manifest write failure + // after moving the images, without adding an I/O fault hook. + conflict := filepath.Join(sandboxManifestName, "fragment") + checkpointDir := ateletpath.CheckpointStateDir(pluginActorUID) + if err := os.MkdirAll(filepath.Join(checkpointDir, sandboxManifestName), 0o700); err != nil { + t.Fatal(err) + } + rec := *a.rec + rec.SnapshotFiles = []string{"checkpoint.img", conflict} + for _, name := range rec.SnapshotFiles { + if err := os.WriteFile(filepath.Join(checkpointDir, name), []byte("uncommitted"), 0o600); err != nil { + t.Fatal(err) + } + } + err := a.s.moveLocalCheckpoint(t.Context(), &ateletpb.CheckpointRequest{ + ActorUid: pluginActorUID, Snapshot: a.snapshot("incomplete-snap", 2), + }, &rec) + if err == nil || !strings.Contains(err.Error(), "writing snapshot manifest") { + t.Fatalf("moveLocalCheckpoint error = %v, want manifest write failure", err) + } + resp, err := a.s.Terminate(t.Context(), &ateletpb.TerminateRequest{ + Atespace: "ate-demo", ActorName: "counter", ActorUid: pluginActorUID, + WorkerPodUid: pluginWorkerPodUID, Spec: a.spec, PreserveSnapshots: true, + }) + if err != nil { + t.Fatal(err) + } + if !proto.Equal(resp.GetState(), want) { + t.Errorf("retained snapshot state = %v, want %v", resp.GetState(), want) + } + if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "incomplete-snap")); !errors.Is(err, os.ErrNotExist) { + t.Errorf("incomplete checkpoint survived Terminate: %v", err) + } + if entries, err := os.ReadDir(checkpointDir); err != nil || len(entries) != 0 { + t.Errorf("checkpoint working directory after Terminate = %v, %v; want empty", entries, err) + } + if keepCommitted { + a.restoreLocal(t, "committed-snap", 1) + } + }) + } +} + // Terminate holds off new uploads. Delete cancels them, while Revert and failed // teardown let pending uploads continue after the call. func TestTerminateHoldsOffBackgroundUpload(t *testing.T) { diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 8b0b88ec17..a2978ac1a2 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -2022,7 +2022,7 @@ func (s *Atelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) if req.GetPreserveSnapshots() { // Workload teardown must not cancel uploads or remove the checkpoint and // sandbox assets needed to restore the last successful suspend. - if err := resetActorDirs(actorUID); err != nil { + if err := actordirs.ResetActorDirs(actorUID); err != nil { return nil, fmt.Errorf("failed to reset actor directories during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) } if err := s.snapshots.ScheduleDetach(actorUID); err != nil { diff --git a/cmd/kubectl-ate/README.md b/cmd/kubectl-ate/README.md index 9ed2f37231..fa53ed2b70 100644 --- a/cmd/kubectl-ate/README.md +++ b/cmd/kubectl-ate/README.md @@ -215,7 +215,7 @@ kubectl ate suspend actor my-actor -a kubectl ate suspend actor my-actor -a --mode=sync # Revert a running or crashed actor to its latest committed snapshot, including a local snapshot whose upload is pending -# A suspended actor without a node-local snapshot is already at its external snapshot, so it cannot be reverted +# A suspended actor cannot be reverted, regardless of snapshot locality kubectl ate revert actor my-actor -a # Delete an actor, terminating workloads and detaching volumes. diff --git a/internal/proto/ateletpb/atelet.pb.go b/internal/proto/ateletpb/atelet.pb.go index 70f40de83c..e869666ce7 100644 --- a/internal/proto/ateletpb/atelet.pb.go +++ b/internal/proto/ateletpb/atelet.pb.go @@ -1833,9 +1833,14 @@ type TerminateRequest struct { // actors with a fallback spec. // // +k8s:optional - Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"` + // Keep committed local snapshots and their background uploads after stopping + // the workload. Revert uses these snapshots for the next resume. + // + // +k8s:optional + PreserveSnapshots bool `protobuf:"varint,8,opt,name=preserve_snapshots,json=preserveSnapshots,proto3" json:"preserve_snapshots,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *TerminateRequest) Reset() { @@ -1917,8 +1922,19 @@ func (x *TerminateRequest) GetSpec() *WorkloadSpec { return nil } +func (x *TerminateRequest) GetPreserveSnapshots() bool { + if x != nil { + return x.PreserveSnapshots + } + return false +} + type TerminateResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` + state protoimpl.MessageState `protogen:"open.v1"` + // Snapshot state retained when preserve_snapshots is set. + // + // +k8s:optional + State *ActorSnapshotState `protobuf:"bytes,1,opt,name=state,proto3" json:"state,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1953,6 +1969,13 @@ func (*TerminateResponse) Descriptor() ([]byte, []int) { return file_internal_proto_ateletpb_atelet_proto_rawDescGZIP(), []int{28} } +func (x *TerminateResponse) GetState() *ActorSnapshotState { + if x != nil { + return x.State + } + return nil +} + type RunRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // The UID of the worker pod whose the worker hosts the actor. @@ -4428,7 +4451,7 @@ const file_internal_proto_ateletpb_atelet_proto_rawDesc = "" + "\tactor_uid\x18\x03 \x01(\tR\bactorUid\x12>\n" + "\x1bcertificate_signing_request\x18\x04 \x01(\fR\x19certificateSigningRequest\"M\n" + "\x1cMintActorCertificateResponse\x12-\n" + - "\x12actor_certificates\x18\x01 \x03(\fR\x11actorCertificates\"\xa2\x02\n" + + "\x12actor_certificates\x18\x01 \x03(\fR\x11actorCertificates\"\xd1\x02\n" + "\x10TerminateRequest\x12$\n" + "\x0eworker_pod_uid\x18\x01 \x01(\tR\fworkerPodUid\x12\x1a\n" + "\batespace\x18\x02 \x01(\tR\batespace\x12\x1d\n" + @@ -4437,8 +4460,10 @@ const file_internal_proto_ateletpb_atelet_proto_rawDesc = "" + "\tactor_uid\x18\x04 \x01(\tR\bactorUid\x126\n" + "\x17actor_template_atespace\x18\x05 \x01(\tR\x15actorTemplateAtespace\x12.\n" + "\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12(\n" + - "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\"\x13\n" + - "\x11TerminateResponse\"\xf4\x03\n" + + "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\x12-\n" + + "\x12preserve_snapshots\x18\b \x01(\bR\x11preserveSnapshots\"E\n" + + "\x11TerminateResponse\x120\n" + + "\x05state\x18\x01 \x01(\v2\x1a.atelet.ActorSnapshotStateR\x05state\"\xf4\x03\n" + "\n" + "RunRequest\x12$\n" + "\x0eworker_pod_uid\x18\x01 \x01(\tR\fworkerPodUid\x12\x1a\n" + @@ -4733,84 +4758,85 @@ var file_internal_proto_ateletpb_atelet_proto_depIdxs = []int32{ 27, // 13: atelet.EgressDestinations.ports:type_name -> atelet.EgressPorts 28, // 14: atelet.EgressPorts.all:type_name -> atelet.EgressAllPorts 35, // 15: atelet.TerminateRequest.spec:type_name -> atelet.WorkloadSpec - 35, // 16: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec - 34, // 17: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway - 15, // 18: atelet.RunRequest.resources:type_name -> atelet.ResourceRequirements - 11, // 19: atelet.RunRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime - 46, // 20: atelet.WorkloadSpec.containers:type_name -> atelet.Container - 44, // 21: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume - 66, // 22: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry - 67, // 23: atelet.ExternalVolumeSource.publish_context:type_name -> atelet.ExternalVolumeSource.PublishContextEntry - 0, // 24: atelet.ActorMetadataItem.field:type_name -> atelet.ActorMetadataField - 39, // 25: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem - 40, // 26: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource - 41, // 27: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource - 42, // 28: atelet.SystemInfoVolumeSource.data_sources:type_name -> atelet.SystemInfoDataSource - 36, // 29: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolumeSource - 37, // 30: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource - 43, // 31: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolumeSource - 38, // 32: atelet.Volume.image:type_name -> atelet.ImageVolumeSource - 49, // 33: atelet.Container.env:type_name -> atelet.EnvVar - 50, // 34: atelet.Container.wakeup_probe:type_name -> atelet.WakeupProbe - 45, // 35: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount - 47, // 36: atelet.Container.security_context:type_name -> atelet.SecurityContext - 15, // 37: atelet.Container.resources:type_name -> atelet.ResourceRequirements - 48, // 38: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities - 51, // 39: atelet.WakeupProbe.http_get:type_name -> atelet.HttpGetAction - 2, // 40: atelet.ObjectStorage.sharing:type_name -> atelet.SnapshotSharing - 1, // 41: atelet.Snapshot.fidelity:type_name -> atelet.SnapshotFidelity - 53, // 42: atelet.Snapshot.object:type_name -> atelet.ObjectStorage - 54, // 43: atelet.SnapshotReport.snapshot:type_name -> atelet.Snapshot - 68, // 44: atelet.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp - 68, // 45: atelet.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp - 55, // 46: atelet.ActorSnapshotState.latest_local:type_name -> atelet.SnapshotReport - 55, // 47: atelet.ActorSnapshotState.uploading:type_name -> atelet.SnapshotReport - 55, // 48: atelet.ActorSnapshotState.latest_durable:type_name -> atelet.SnapshotReport - 68, // 49: atelet.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp - 3, // 50: atelet.UploadConfig.mode:type_name -> atelet.UploadMode - 35, // 51: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec - 54, // 52: atelet.CheckpointRequest.snapshot:type_name -> atelet.Snapshot - 57, // 53: atelet.CheckpointRequest.upload_config:type_name -> atelet.UploadConfig - 56, // 54: atelet.CheckpointResponse.state:type_name -> atelet.ActorSnapshotState - 35, // 55: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec - 54, // 56: atelet.RestoreRequest.snapshot:type_name -> atelet.Snapshot - 34, // 57: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway - 15, // 58: atelet.RestoreRequest.resources:type_name -> atelet.ResourceRequirements - 11, // 59: atelet.RestoreRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime - 54, // 60: atelet.PromoteSnapshotRequest.snapshot:type_name -> atelet.Snapshot - 56, // 61: atelet.PromoteSnapshotResponse.state:type_name -> atelet.ActorSnapshotState - 56, // 62: atelet.DetachActorResponse.state:type_name -> atelet.ActorSnapshotState - 29, // 63: atelet.WorkerSupport.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest - 4, // 64: atelet.WorkerSupport.AcquireWorkerEpoch:input_type -> atelet.AcquireWorkerEpochRequest - 6, // 65: atelet.WorkerSupport.RegisterWorker:input_type -> atelet.RegisterWorkerRequest - 7, // 66: atelet.WorkerSupport.SetWorkerSandboxRuntimes:input_type -> atelet.SetWorkerSandboxRuntimesRequest - 18, // 67: atelet.WorkerSupport.RequestActorSuspend:input_type -> atelet.RequestActorSuspendRequest - 22, // 68: atelet.WorkerSupport.GetActorEgressPolicy:input_type -> atelet.GetActorEgressPolicyRequest - 20, // 69: atelet.WorkerSupport.ReportActorCrashed:input_type -> atelet.ReportActorCrashedRequest - 33, // 70: atelet.Atelet.Run:input_type -> atelet.RunRequest - 58, // 71: atelet.Atelet.Checkpoint:input_type -> atelet.CheckpointRequest - 60, // 72: atelet.Atelet.Restore:input_type -> atelet.RestoreRequest - 62, // 73: atelet.Atelet.PromoteSnapshot:input_type -> atelet.PromoteSnapshotRequest - 64, // 74: atelet.Atelet.DetachActor:input_type -> atelet.DetachActorRequest - 31, // 75: atelet.Atelet.Terminate:input_type -> atelet.TerminateRequest - 30, // 76: atelet.WorkerSupport.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse - 5, // 77: atelet.WorkerSupport.AcquireWorkerEpoch:output_type -> atelet.AcquireWorkerEpochResponse - 17, // 78: atelet.WorkerSupport.RegisterWorker:output_type -> atelet.RegisterWorkerResponse - 8, // 79: atelet.WorkerSupport.SetWorkerSandboxRuntimes:output_type -> atelet.SetWorkerSandboxRuntimesResponse - 19, // 80: atelet.WorkerSupport.RequestActorSuspend:output_type -> atelet.RequestActorSuspendResponse - 23, // 81: atelet.WorkerSupport.GetActorEgressPolicy:output_type -> atelet.GetActorEgressPolicyResponse - 21, // 82: atelet.WorkerSupport.ReportActorCrashed:output_type -> atelet.ReportActorCrashedResponse - 52, // 83: atelet.Atelet.Run:output_type -> atelet.RunResponse - 59, // 84: atelet.Atelet.Checkpoint:output_type -> atelet.CheckpointResponse - 61, // 85: atelet.Atelet.Restore:output_type -> atelet.RestoreResponse - 63, // 86: atelet.Atelet.PromoteSnapshot:output_type -> atelet.PromoteSnapshotResponse - 65, // 87: atelet.Atelet.DetachActor:output_type -> atelet.DetachActorResponse - 32, // 88: atelet.Atelet.Terminate:output_type -> atelet.TerminateResponse - 76, // [76:89] is the sub-list for method output_type - 63, // [63:76] is the sub-list for method input_type - 63, // [63:63] is the sub-list for extension type_name - 63, // [63:63] is the sub-list for extension extendee - 0, // [0:63] is the sub-list for field type_name + 56, // 16: atelet.TerminateResponse.state:type_name -> atelet.ActorSnapshotState + 35, // 17: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec + 34, // 18: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway + 15, // 19: atelet.RunRequest.resources:type_name -> atelet.ResourceRequirements + 11, // 20: atelet.RunRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime + 46, // 21: atelet.WorkloadSpec.containers:type_name -> atelet.Container + 44, // 22: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume + 66, // 23: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry + 67, // 24: atelet.ExternalVolumeSource.publish_context:type_name -> atelet.ExternalVolumeSource.PublishContextEntry + 0, // 25: atelet.ActorMetadataItem.field:type_name -> atelet.ActorMetadataField + 39, // 26: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem + 40, // 27: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource + 41, // 28: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource + 42, // 29: atelet.SystemInfoVolumeSource.data_sources:type_name -> atelet.SystemInfoDataSource + 36, // 30: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolumeSource + 37, // 31: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource + 43, // 32: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolumeSource + 38, // 33: atelet.Volume.image:type_name -> atelet.ImageVolumeSource + 49, // 34: atelet.Container.env:type_name -> atelet.EnvVar + 50, // 35: atelet.Container.wakeup_probe:type_name -> atelet.WakeupProbe + 45, // 36: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount + 47, // 37: atelet.Container.security_context:type_name -> atelet.SecurityContext + 15, // 38: atelet.Container.resources:type_name -> atelet.ResourceRequirements + 48, // 39: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities + 51, // 40: atelet.WakeupProbe.http_get:type_name -> atelet.HttpGetAction + 2, // 41: atelet.ObjectStorage.sharing:type_name -> atelet.SnapshotSharing + 1, // 42: atelet.Snapshot.fidelity:type_name -> atelet.SnapshotFidelity + 53, // 43: atelet.Snapshot.object:type_name -> atelet.ObjectStorage + 54, // 44: atelet.SnapshotReport.snapshot:type_name -> atelet.Snapshot + 68, // 45: atelet.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp + 68, // 46: atelet.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp + 55, // 47: atelet.ActorSnapshotState.latest_local:type_name -> atelet.SnapshotReport + 55, // 48: atelet.ActorSnapshotState.uploading:type_name -> atelet.SnapshotReport + 55, // 49: atelet.ActorSnapshotState.latest_durable:type_name -> atelet.SnapshotReport + 68, // 50: atelet.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp + 3, // 51: atelet.UploadConfig.mode:type_name -> atelet.UploadMode + 35, // 52: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec + 54, // 53: atelet.CheckpointRequest.snapshot:type_name -> atelet.Snapshot + 57, // 54: atelet.CheckpointRequest.upload_config:type_name -> atelet.UploadConfig + 56, // 55: atelet.CheckpointResponse.state:type_name -> atelet.ActorSnapshotState + 35, // 56: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec + 54, // 57: atelet.RestoreRequest.snapshot:type_name -> atelet.Snapshot + 34, // 58: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway + 15, // 59: atelet.RestoreRequest.resources:type_name -> atelet.ResourceRequirements + 11, // 60: atelet.RestoreRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime + 54, // 61: atelet.PromoteSnapshotRequest.snapshot:type_name -> atelet.Snapshot + 56, // 62: atelet.PromoteSnapshotResponse.state:type_name -> atelet.ActorSnapshotState + 56, // 63: atelet.DetachActorResponse.state:type_name -> atelet.ActorSnapshotState + 29, // 64: atelet.WorkerSupport.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest + 4, // 65: atelet.WorkerSupport.AcquireWorkerEpoch:input_type -> atelet.AcquireWorkerEpochRequest + 6, // 66: atelet.WorkerSupport.RegisterWorker:input_type -> atelet.RegisterWorkerRequest + 7, // 67: atelet.WorkerSupport.SetWorkerSandboxRuntimes:input_type -> atelet.SetWorkerSandboxRuntimesRequest + 18, // 68: atelet.WorkerSupport.RequestActorSuspend:input_type -> atelet.RequestActorSuspendRequest + 22, // 69: atelet.WorkerSupport.GetActorEgressPolicy:input_type -> atelet.GetActorEgressPolicyRequest + 20, // 70: atelet.WorkerSupport.ReportActorCrashed:input_type -> atelet.ReportActorCrashedRequest + 33, // 71: atelet.Atelet.Run:input_type -> atelet.RunRequest + 58, // 72: atelet.Atelet.Checkpoint:input_type -> atelet.CheckpointRequest + 60, // 73: atelet.Atelet.Restore:input_type -> atelet.RestoreRequest + 62, // 74: atelet.Atelet.PromoteSnapshot:input_type -> atelet.PromoteSnapshotRequest + 64, // 75: atelet.Atelet.DetachActor:input_type -> atelet.DetachActorRequest + 31, // 76: atelet.Atelet.Terminate:input_type -> atelet.TerminateRequest + 30, // 77: atelet.WorkerSupport.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse + 5, // 78: atelet.WorkerSupport.AcquireWorkerEpoch:output_type -> atelet.AcquireWorkerEpochResponse + 17, // 79: atelet.WorkerSupport.RegisterWorker:output_type -> atelet.RegisterWorkerResponse + 8, // 80: atelet.WorkerSupport.SetWorkerSandboxRuntimes:output_type -> atelet.SetWorkerSandboxRuntimesResponse + 19, // 81: atelet.WorkerSupport.RequestActorSuspend:output_type -> atelet.RequestActorSuspendResponse + 23, // 82: atelet.WorkerSupport.GetActorEgressPolicy:output_type -> atelet.GetActorEgressPolicyResponse + 21, // 83: atelet.WorkerSupport.ReportActorCrashed:output_type -> atelet.ReportActorCrashedResponse + 52, // 84: atelet.Atelet.Run:output_type -> atelet.RunResponse + 59, // 85: atelet.Atelet.Checkpoint:output_type -> atelet.CheckpointResponse + 61, // 86: atelet.Atelet.Restore:output_type -> atelet.RestoreResponse + 63, // 87: atelet.Atelet.PromoteSnapshot:output_type -> atelet.PromoteSnapshotResponse + 65, // 88: atelet.Atelet.DetachActor:output_type -> atelet.DetachActorResponse + 32, // 89: atelet.Atelet.Terminate:output_type -> atelet.TerminateResponse + 77, // [77:90] is the sub-list for method output_type + 64, // [64:77] is the sub-list for method input_type + 64, // [64:64] is the sub-list for extension type_name + 64, // [64:64] is the sub-list for extension extendee + 0, // [0:64] is the sub-list for field type_name } func init() { file_internal_proto_ateletpb_atelet_proto_init() } From 8678da7efadcbaacca8dcfc80b8cdea620663ca3 Mon Sep 17 00:00:00 2001 From: cloudonly Date: Fri, 23 Oct 2026 16:29:59 +0800 Subject: [PATCH 3/4] Forget deleted snapshot nodes before reverting --- .../internal/controlapi/workflow_revert.go | 41 +++++++- .../controlapi/workflow_revert_test.go | 99 +++++++++++++++++++ 2 files changed, 139 insertions(+), 1 deletion(-) diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go index 057c29ef74..d601d9a9f6 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert.go @@ -182,6 +182,41 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso ctx, done := stepSpan(ctx, "DiscardWorker") defer func() { err = done(err) }() + // A failed restore on another node can leave the snapshot's locality on + // the deleted source node. The current node's Terminate cannot clear it. + checkedNodes := map[string]bool{"": true, actor.GetStatus().GetAssignedNode(): true} + goneNodes := map[string]bool{} + for _, snap := range actor.GetStatus().GetSnapshots() { + node := snap.GetLocality() + if checkedNodes[node] { + continue + } + checkedNodes[node] = true + gone, err := w.dialer.NodeGone(ctx, node) + if err != nil { + return fmt.Errorf("while checking snapshot node %q: %w", node, err) + } + if gone { + goneNodes[node] = true + } + } + if len(goneNodes) > 0 { + actor, err = w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(a *ateapi.Actor) error { + for _, snapshot := range a.GetStatus().GetSnapshots() { + if goneNodes[snapshot.GetLocality()] { + snapshot.Locality = "" + } + } + return nil + }) + if errors.Is(err, store.ErrVersionConflict) { + return apierror.Aborted("concurrent update conflict, please retry") + } + if err != nil { + return fmt.Errorf("while clearing snapshot locality on deleted nodes: %w", err) + } + } + if actor.GetStatus().GetAssignedNode() != "" { forgetNode := false snapshotState, nodeGone, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert, true) @@ -204,7 +239,11 @@ func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef reso // actor is crashed. Reconcile before deleting any pending objects. if _, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(a *ateapi.Actor) error { if nodeGone { - clearLocalSnapshots(a.Status) + for _, snapshot := range a.GetStatus().GetSnapshots() { + if snapshot.GetLocality() == actor.GetStatus().GetAssignedNode() { + snapshot.Locality = "" + } + } return nil } return applyRevertedSnapshotState(ctx, a.Status, actor.GetStatus().GetAssignedNode(), snapshotState) diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go index b3c7dbd2fe..01d7311038 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go @@ -328,6 +328,105 @@ func TestRevertActor_AteletAlreadyDetached(t *testing.T) { } } +func TestRevertActor_SnapshotOnAnotherNode(t *testing.T) { + for _, tt := range []struct { + name string + kept bool + assignedGone bool + lookupErr error + lookupNode string + conflict bool + wantCode codes.Code + }{ + {name: "deleted source node"}, + {name: "source node still exists", kept: true}, + {name: "assigned node deleted, source node still exists", kept: true, assignedGone: true}, + {name: "node lookup fails", lookupErr: apierror.Unavailable("node lookup failed"), wantCode: codes.Unavailable}, + {name: "later node lookup fails", lookupErr: apierror.Unavailable("node lookup failed"), lookupNode: "node-4", wantCode: codes.Unavailable}, + {name: "concurrent update", conflict: true, wantCode: codes.Aborted}, + } { + t.Run(tt.name, func(t *testing.T) { + ctx := t.Context() + st := newTestPersistence(t) + w, atelet := newWireCaptureWorkflow(t, st) + ref := resources.ActorRef{Atespace: "team-a", Name: "id1"} + assignedNode, snapshotNode := "node-1", "node-3" + if tt.assignedGone { + assignedNode, snapshotNode = snapshotNode, assignedNode + } + seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", ateapi.ActorState_ACTOR_STATE_REVERTING, func(a *ateapi.Actor) { + a.Status.AssignedNode = assignedNode + a.Status.LastAssignedGeneration = 2 + a.Status.Snapshots = []*ateapi.Snapshot{ + newPausedSnapshot(t, 1, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "older-pending", snapshotNode), + newPausedSnapshot(t, 2, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "pending", snapshotNode), + } + if tt.lookupNode != "" { + a.Status.Snapshots[1].Locality = tt.lookupNode + } + }) + actor, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatal(err) + } + want := proto.CloneOf(actor.GetStatus()).GetSnapshots() + var nodes []string + if tt.kept { + nodes = append(nodes, snapshotNode) + } + lookup := nodeGetter(nil, nodes...) + lookups := map[string]int{} + WithNodeGetter(func(ctx context.Context, node string) error { + lookups[node]++ + if tt.lookupErr != nil && (tt.lookupNode == "" || tt.lookupNode == node) { + return tt.lookupErr + } + if tt.conflict { + if _, err := st.UpdateActor(ctx, ref, store.PreconditionFrom(actor), func(a *ateapi.Actor) error { + a.WorkerSelector = &ateapi.Selector{MatchLabels: map[string]string{"changed": "true"}} + return nil + }); err != nil { + t.Fatal(err) + } + } + return lookup(ctx, node) + })(w.dialer) + + err = w.ensureWorkerDiscarded(ctx, ref, actor, nil) + if apierror.Code(err) != tt.wantCode { + t.Fatalf("ensureWorkerDiscarded = %v, want %v", err, tt.wantCode) + } + stored, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatal(err) + } + if !tt.kept && tt.wantCode == codes.OK { + for _, snapshot := range want { + snapshot.Locality = "" + } + } + got := stored.GetStatus().GetSnapshots() + if len(got) != len(want) { + t.Fatalf("snapshots = %v, want retained pending snapshots %v", got, want) + } + for i := range want { + if !proto.Equal(got[i], want[i]) { + t.Errorf("snapshot = %v, want retained pending snapshot %v", got[i], want[i]) + } + } + if lookups[snapshotNode] != 1 { + t.Errorf("snapshot node lookups = %d, want 1", lookups[snapshotNode]) + } + if tt.conflict && stored.GetWorkerSelector().GetMatchLabels()["changed"] != "true" { + t.Error("concurrent update was overwritten") + } + if tt.wantCode != codes.OK && atelet.terminateRequest() != nil { + t.Error("terminated the actor after failing to check the snapshot node") + } + }) + } +} + func TestRevertActor_UnreachableNode(t *testing.T) { for _, tt := range []struct { name string From 40dc5fc28ee516a77e6aa3be239a36b8223ec19f Mon Sep 17 00:00:00 2001 From: cloudonly Date: Sat, 24 Oct 2026 22:24:48 +0800 Subject: [PATCH 4/4] Keep local snapshots when DeleteWorker reclaims an actor's node state DeleteWorker sent a full Terminate for each actor on a deleted worker, which forgot the actor's committed local snapshot and pruned its checkpoint. A later Revert could not account for that snapshot and failed on every retry. Send preserve_snapshots, as Revert does, so the snapshot and its pending upload survive until the revert. --- cmd/ateapi/internal/controlapi/workflow_worker_delete.go | 7 ++++--- .../internal/controlapi/workflow_worker_delete_test.go | 7 ++++++- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go index 3e706c5968..89e1b20df6 100644 --- a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go @@ -300,9 +300,9 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapi.Wor } // reclaimActorStateOnNode asks the atelet on the worker's node to terminate the -// actor, removing /var/lib/ate/actors/ and /var/lib/ate/snapshots/ -// (durable dir, checkpoint and restore images, and local snapshots) and -// unmounting its external volumes. +// actor, emptying /var/lib/ate/actors/ and unmounting its external +// volumes. The actor's committed local snapshot and its pending upload stay on +// the node, so the actor's revert can return to them. // // The worker's pod and process are gone by the time this runs, while the // node's atelet remains and reclaims the directories without a live worker. @@ -353,6 +353,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at ActorTemplateAtespace: actor.GetActorTemplate().GetAtespace(), ActorTemplateName: actor.GetActorTemplate().GetName(), Spec: fallbackWorkloadSpec(actor), + PreserveSnapshots: true, }) switch status.Code(err) { case codes.OK: diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go index aeed776291..4262451fe8 100644 --- a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go @@ -448,7 +448,7 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) { { // An actor resumed from a local snapshot keeps Locality on that // snapshot while running. When its worker pod vanishes, it crashes - // and its node state is reclaimed along with any local snapshot. + // and its node state is reclaimed, but not the local snapshot. name: "running actor with a local snapshot is reclaimed", state: ateapi.ActorState_ACTOR_STATE_RUNNING, seed: withPausedShape(t, "node-1"), @@ -515,6 +515,11 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) { if want := validWorker(apiWorkerName).GetWorkerPodUid(); req.GetWorkerPodUid() != want { t.Errorf("Terminate worker_pod_uid = %q, want the worker's pod uid %q", req.GetWorkerPodUid(), want) } + // A revert restores from the crashed actor's committed local + // snapshot. Reclaiming it here would leave that revert stuck. + if !req.GetPreserveSnapshots() { + t.Error("Terminate preserve_snapshots = false, want true") + } }) } }