diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 0fcd67c02..d1179f8d4 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -508,6 +508,31 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + // A checkpoint whose snapshot is already at its destination is done, and + // re-running it would drive a sandbox the first attempt destroyed (#372). + // The control plane mints the destination once per suspend/pause and + // re-sends it on every re-entry, so a manifest there names THIS checkpoint + // and no other. + // + // Checked before the metrics defer below is installed: a replay that does + // no work is not a checkpoint, and recording it as one would report a + // near-zero duration against snapshots that take seconds to write. + // Checked before pruneLocalCheckpoints too, which would otherwise delete + // the local snapshot that proves the earlier success. + // + // Costs one small object read per external checkpoint. That is paid before + // the guest is paused, against an operation that goes on to move + // gigabytes. + committed, err := s.checkpointAlreadyCommitted(ctx, req) + if err != nil { + return nil, err + } + if committed { + slog.InfoContext(ctx, "Checkpoint already committed to its destination; nothing to do", + slog.Any("actor", actorRef), slog.String("actor_uid", actorUID)) + return &ateletpb.CheckpointResponse{}, nil + } + // Per-phase timing, recorded on the way out so a failed checkpoint still // reports the phases it completed. Phases left at zero never ran. tStart := time.Now() @@ -638,6 +663,62 @@ func toAteomSnapshotScope(scope ateletpb.SnapshotScope) ateompb.SnapshotScope { } } +// checkpointAlreadyCommitted reports whether the snapshot this request asks +// for is already written to its destination, i.e. whether an earlier attempt +// at this same checkpoint completed and only its response went missing. +// +// The manifest is the commit marker on both paths, which is what makes this +// answer trustworthy: uploadSnapshot writes it last and never in parallel with +// the files it lists, and moveLocalCheckpoint writes it after the last rename. +// A manifest therefore implies every file it names is already in place, while +// an interrupted attempt leaves at most orphaned files and no manifest. +func (s *AteomHerder) checkpointAlreadyCommitted(ctx context.Context, req *ateletpb.CheckpointRequest) (bool, error) { + switch req.GetType() { + case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL: + uri, err := resources.ParseSnapshotURI(req.GetExternalConfig().GetSnapshotUri()) + if err != nil { + return false, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidObjectURL) + } + return s.snapshotManifestUploaded(ctx, uri) + + case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL: + path := filepath.Join(ateompath.LocalSnapshotDir(req.GetActorUid(), req.GetLocalConfig().GetSnapshotName()), sandboxManifestName) + switch _, err := os.Stat(path); { + case err == nil: + return true, nil + case errors.Is(err, os.ErrNotExist): + return false, nil + default: + return false, wrapFileSystemErr("while probing for an already-written local snapshot manifest", err) + } + + default: + // Unreachable: validateCheckpointRequest rejects any other type before + // this runs. Answering "not committed" leaves the rejection to the + // type switches that own it rather than inventing a second message. + return false, nil + } +} + +// snapshotManifestUploaded reports whether the snapshot at uri has its +// manifest in object storage. A missing manifest is an answer, not a failure; +// any other error is one, and is returned rather than read as "not there" — +// treating an unreachable bucket as "not committed" would re-run a destructive +// checkpoint on the strength of a failed lookup. +func (s *AteomHerder) snapshotManifestUploaded(ctx context.Context, uri resources.SnapshotURI) (bool, error) { + manifestURI, err := uri.ObjectURI(sandboxManifestName) + if err != nil { + return false, ateerrors.CrashIfReason(ctx, fmt.Errorf("while addressing snapshot manifest in GCS: %w", err), ateerrors.ReasonInvalidObjectURL) + } + if _, err := ategcs.FetchFromGCS(ctx, s.gcsClient, manifestURI); err != nil { + if errors.Is(err, ateerrors.ReasonFailedGetExternalObject) { + return false, nil + } + return false, fmt.Errorf("while probing for an already-uploaded snapshot manifest: %w", err) + } + return true, nil +} + func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.CheckpointRequest, checkpointDir string, rec *sandboxAssetsRecord) error { localCheckpointPath := ateompath.LocalSnapshotDir(req.GetActorUid(), req.GetLocalConfig().GetSnapshotName()) if err := os.MkdirAll(localCheckpointPath, 0o700); err != nil { @@ -645,12 +726,26 @@ func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.Che } // Move exactly the files ateom reported. + // + // A file already at the destination with nothing left at the source was + // moved by an earlier attempt at this same checkpoint: the rename is not + // repeatable, so re-entry has to recognize its own work rather than fail on + // the missing source. Only the manifest below commits the snapshot, so a + // half-moved set is exactly what an interrupted attempt leaves behind. for _, fileName := range rec.SnapshotFiles { src := filepath.Join(checkpointDir, fileName) dst := filepath.Join(localCheckpointPath, fileName) recordSnapshotSize(ctx, fileName, src, req.GetActorTemplateNamespace(), req.GetActorTemplateName()) - if err := os.Rename(src, dst); err != nil { + err := os.Rename(src, dst) + if errors.Is(err, os.ErrNotExist) { + if _, statErr := os.Stat(dst); statErr == nil { + continue + } + // Gone from both sides: the snapshot cannot be assembled. + return wrapFileSystemErr(fmt.Sprintf("snapshot file %q is missing from both %s and %s", fileName, checkpointDir, localCheckpointPath), err) + } + if err != nil { return fmt.Errorf("failed to move %s to %s: %w", src, dst, err) } } @@ -767,11 +862,6 @@ func (s *AteomHerder) UploadPausedCheckpoint(ctx context.Context, req *ateletpb. // returns the sandbox class recorded in the snapshot manifest (empty when the // manifest was not read). Parameterized by localDir for tests. func (s *AteomHerder) uploadLocalCheckpointDir(ctx context.Context, req *ateletpb.UploadPausedCheckpointRequest, localDir string, uri resources.SnapshotURI) (string, error) { - manifestURI, err := uri.ObjectURI(sandboxManifestName) - if err != nil { - return "", fmt.Errorf("while addressing snapshot manifest in GCS: %w", err) - } - manifest, err := os.ReadFile(filepath.Join(localDir, sandboxManifestName)) if errors.Is(err, os.ErrNotExist) { // The local snapshot is gone. A previous invocation may have uploaded @@ -779,16 +869,16 @@ func (s *AteomHerder) uploadLocalCheckpointDir(ctx context.Context, req *ateletp // means the whole snapshot is committed and this retry already // succeeded. Absent on both sides, the paused actor's state is // unrecoverable. - _, fetchErr := ategcs.FetchFromGCS(ctx, s.gcsClient, manifestURI) - if fetchErr == nil { + uploaded, probeErr := s.snapshotManifestUploaded(ctx, uri) + if probeErr != nil { + return "", probeErr + } + if uploaded { slog.InfoContext(ctx, "Local snapshot already uploaded and pruned; nothing to do", slog.String("snapshot_uri", req.GetDestinationSnapshotUri())) return "", nil } - if errors.Is(fetchErr, ateerrors.ReasonFailedGetExternalObject) { - return "", ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonLocalSnapshotGone, ateerrors.ActorCrashedMetadata(), - fmt.Errorf("local snapshot %q is gone and no uploaded copy exists: %w", req.GetLocalSnapshotName(), fetchErr)) - } - return "", fmt.Errorf("while probing for an already-uploaded snapshot manifest: %w", fetchErr) + return "", ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonLocalSnapshotGone, ateerrors.ActorCrashedMetadata(), + fmt.Errorf("%w: local snapshot %q is gone and no uploaded copy exists", ateerrors.ReasonLocalSnapshotGone, req.GetLocalSnapshotName())) } if err != nil { return "", wrapFileSystemErr("while reading local snapshot manifest", err) diff --git a/cmd/atelet/main_test.go b/cmd/atelet/main_test.go index 1bb492db4..423582e77 100644 --- a/cmd/atelet/main_test.go +++ b/cmd/atelet/main_test.go @@ -1253,11 +1253,17 @@ type recordingObjectStorage struct { mu sync.Mutex objects map[string][]byte putErr error + // getErr stands in for a storage backend that is unreachable rather than + // empty — an error a caller must not read as "the object is not there". + getErr error } func (r *recordingObjectStorage) GetObject(_ context.Context, bucket, object string) (io.ReadCloser, error) { r.mu.Lock() defer r.mu.Unlock() + if r.getErr != nil { + return nil, r.getErr + } b, ok := r.objects[bucket+"/"+object] if !ok { return nil, fmt.Errorf("%w: Bucket:%q, Object:%q", ateerrors.ReasonFailedGetExternalObject, bucket, object) @@ -1580,3 +1586,179 @@ func TestValidateUploadPausedCheckpointRequest(t *testing.T) { }) } } + +// useTempActorsDir points the shared actor-state root at a temp directory, so +// tests touching an actor's on-node paths (local snapshots, checkpoint state) +// stay off the node's real /var/lib tree. +func useTempActorsDir(t *testing.T) { + t.Helper() + orig := ateompath.ActorsDir + t.Cleanup(func() { ateompath.ActorsDir = orig }) + ateompath.ActorsDir = t.TempDir() +} + +func TestCheckpointAlreadyCommitted(t *testing.T) { + ctx := context.Background() + const manifestKey = "bucket/root/snapshots/ate-demo/counter-1-snap/manifest.json" + + t.Run("external with an uploaded manifest", func(t *testing.T) { + s := &AteomHerder{gcsClient: &recordingObjectStorage{ + objects: map[string][]byte{manifestKey: []byte(`{"pauseImage":"pause:v1"}`)}, + }} + + got, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest()) + if err != nil { + t.Fatalf("checkpointAlreadyCommitted: %v", err) + } + if !got { + t.Error("committed = false, want true: the manifest is the commit marker") + } + }) + + t.Run("external with no manifest", func(t *testing.T) { + s := &AteomHerder{gcsClient: &recordingObjectStorage{}} + + got, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest()) + if err != nil { + t.Fatalf("checkpointAlreadyCommitted: %v", err) + } + if got { + t.Error("committed = true, want false") + } + }) + + t.Run("external probe failure is not read as uncommitted", func(t *testing.T) { + // Reading an unreachable bucket as "not committed" would send a + // destructive checkpoint down the re-run path on the strength of a + // failed lookup. + s := &AteomHerder{gcsClient: &recordingObjectStorage{getErr: errors.New("bucket unreachable")}} + + if _, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest()); err == nil { + t.Fatal("checkpointAlreadyCommitted succeeded, want the probe error surfaced") + } + }) + + t.Run("local with a written manifest", func(t *testing.T) { + useTempActorsDir(t) + req := validCheckpointRequest() + req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL + req.Config = &ateletpb.CheckpointRequest_LocalConfig{ + LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"}, + } + writeLocalSnapshot(t, ateompath.LocalSnapshotDir(req.GetActorUid(), "pause-snap-1"), + sandboxAssetsRecord{SandboxClass: "gvisor", PauseImage: testPauseImage, SnapshotFiles: []string{"checkpoint.img"}}, + map[string]string{"checkpoint.img": "img"}) + + got, err := (&AteomHerder{}).checkpointAlreadyCommitted(ctx, req) + if err != nil { + t.Fatalf("checkpointAlreadyCommitted: %v", err) + } + if !got { + t.Error("committed = false, want true") + } + }) + + t.Run("local with no snapshot dir", func(t *testing.T) { + useTempActorsDir(t) + req := validCheckpointRequest() + req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL + req.Config = &ateletpb.CheckpointRequest_LocalConfig{ + LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"}, + } + + got, err := (&AteomHerder{}).checkpointAlreadyCommitted(ctx, req) + if err != nil { + t.Fatalf("checkpointAlreadyCommitted: %v", err) + } + if got { + t.Error("committed = true, want false") + } + }) +} + +func TestCheckpointFastForwardsWhenAlreadyCommitted(t *testing.T) { + // No sandbox record on disk and no ateom dialer: every step after the + // commit probe would fail, so a success here can only come from the + // fast-forward. + useTempActorsDir(t) + s := &AteomHerder{gcsClient: &recordingObjectStorage{ + objects: map[string][]byte{ + "bucket/root/snapshots/ate-demo/counter-1-snap/manifest.json": []byte(`{"pauseImage":"pause:v1"}`), + }, + }} + + resp, err := s.Checkpoint(context.Background(), validCheckpointRequest()) + if err != nil { + t.Fatalf("Checkpoint: %v", err) + } + if resp == nil { + t.Fatal("Checkpoint returned a nil response") + } +} + +func TestMoveLocalCheckpointResumesPartialMove(t *testing.T) { + ctx := context.Background() + useTempActorsDir(t) + + req := validCheckpointRequest() + req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL + req.Config = &ateletpb.CheckpointRequest_LocalConfig{ + LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"}, + } + rec := &sandboxAssetsRecord{ + SandboxClass: "gvisor", + PauseImage: testPauseImage, + SnapshotFiles: []string{"checkpoint.img", "pages.img"}, + } + + // The state an interrupted move leaves: one file already renamed into the + // snapshot dir, the other still in the checkpoint dir, no manifest. + checkpointDir := ateompath.CheckpointStateDir(req.GetActorUid()) + dstDir := ateompath.LocalSnapshotDir(req.GetActorUid(), "pause-snap-1") + for dir, files := range map[string]map[string]string{ + checkpointDir: {"pages.img": "pages"}, + dstDir: {"checkpoint.img": "img"}, + } { + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatalf("creating %s: %v", dir, err) + } + for name, body := range files { + if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil { + t.Fatalf("writing %s: %v", name, err) + } + } + } + + if err := (&AteomHerder{}).moveLocalCheckpoint(ctx, req, checkpointDir, rec); err != nil { + t.Fatalf("moveLocalCheckpoint: %v", err) + } + + for _, name := range append(rec.SnapshotFiles, sandboxManifestName) { + if _, err := os.Stat(filepath.Join(dstDir, name)); err != nil { + t.Errorf("%s missing from the snapshot dir: %v", name, err) + } + } +} + +func TestMoveLocalCheckpointFailsWhenFileGoneFromBothSides(t *testing.T) { + useTempActorsDir(t) + + req := validCheckpointRequest() + req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL + req.Config = &ateletpb.CheckpointRequest_LocalConfig{ + LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"}, + } + checkpointDir := ateompath.CheckpointStateDir(req.GetActorUid()) + if err := os.MkdirAll(checkpointDir, 0o700); err != nil { + t.Fatalf("creating checkpoint dir: %v", err) + } + + rec := &sandboxAssetsRecord{SandboxClass: "gvisor", PauseImage: testPauseImage, SnapshotFiles: []string{"checkpoint.img"}} + err := (&AteomHerder{}).moveLocalCheckpoint(context.Background(), req, checkpointDir, rec) + if err == nil { + t.Fatal("moveLocalCheckpoint succeeded, want a failure: the snapshot cannot be assembled") + } + if !errors.Is(err, ateerrors.ReasonTerminalFileSystemError) { + t.Errorf("err = %v, want it tagged %v", err, ateerrors.ReasonTerminalFileSystemError) + } +} diff --git a/cmd/ateom-gvisor/checkpoint_test.go b/cmd/ateom-gvisor/checkpoint_test.go new file mode 100644 index 000000000..595c12138 --- /dev/null +++ b/cmd/ateom-gvisor/checkpoint_test.go @@ -0,0 +1,95 @@ +//go:build linux + +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "os" + "path/filepath" + "slices" + "testing" + + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/checkpointmarker" + "github.com/agent-substrate/substrate/internal/proto/ateompb" +) + +// useTempActorsDir points the shared actor-state root at a temp directory and +// creates the actor's checkpoint dir. +func useTempActorsDir(t *testing.T, actorUID string) string { + t.Helper() + orig := ateompath.ActorsDir + t.Cleanup(func() { ateompath.ActorsDir = orig }) + ateompath.ActorsDir = t.TempDir() + + dir := ateompath.CheckpointStateDir(actorUID) + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatalf("creating checkpoint dir: %v", err) + } + return dir +} + +func TestListSnapshotFilesExcludesCompletionMarker(t *testing.T) { + const actorUID = "actor-1" + dir := useTempActorsDir(t, actorUID) + + for _, name := range []string{"checkpoint.img", "pages.img"} { + if err := os.WriteFile(filepath.Join(dir, name), []byte("x"), 0o600); err != nil { + t.Fatalf("writing %s: %v", name, err) + } + } + if err := checkpointmarker.Write(actorUID, []string{"checkpoint.img", "pages.img"}); err != nil { + t.Fatalf("checkpointmarker.Write: %v", err) + } + + got, err := listSnapshotFiles(dir) + if err != nil { + t.Fatalf("listSnapshotFiles: %v", err) + } + // The marker is ateom's bookkeeping. Shipping it as snapshot content would + // put it in the manifest, and a restore would then expect it back. + want := []string{"checkpoint.img", "pages.img"} + if !slices.Equal(got, want) { + t.Errorf("listSnapshotFiles = %v, want %v", got, want) + } +} + +func TestCheckpointWorkloadReplaysCompletedCheckpoint(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + want := []string{"checkpoint.img", "pages.img"} + if err := checkpointmarker.Write(actorUID, want); err != nil { + t.Fatalf("checkpointmarker.Write: %v", err) + } + + // A zero-value service with no sandbox and no runsc path: reaching the + // checkpoint itself would fail, so a success here can only be the replay. + s := &AteomService{} + resp, err := s.CheckpointWorkload(context.Background(), &ateompb.CheckpointWorkloadRequest{ + Atespace: "ate-demo", + ActorName: "counter-1", + ActorUid: actorUID, + Scope: ateompb.SnapshotScope_SNAPSHOT_SCOPE_FULL, + }) + if err != nil { + t.Fatalf("CheckpointWorkload: %v", err) + } + if !slices.Equal(resp.GetSnapshotFiles(), want) { + t.Errorf("SnapshotFiles = %v, want %v (the recorded result, replayed verbatim)", resp.GetSnapshotFiles(), want) + } +} diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index f65aa9356..06211595d 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -35,11 +35,13 @@ import ( "cloud.google.com/go/compute/metadata" "github.com/agent-substrate/substrate/internal/actorlog" + "github.com/agent-substrate/substrate/internal/ateerrors" "github.com/agent-substrate/substrate/internal/ateinterceptors" "github.com/agent-substrate/substrate/internal/ateomnet" "github.com/agent-substrate/substrate/internal/ateompath" "github.com/agent-substrate/substrate/internal/ateomstats" "github.com/agent-substrate/substrate/internal/atunnel" + "github.com/agent-substrate/substrate/internal/checkpointmarker" "github.com/agent-substrate/substrate/internal/contextlogging" "github.com/agent-substrate/substrate/internal/imagecache" "github.com/agent-substrate/substrate/internal/proto/ateompb" @@ -679,11 +681,27 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec s.setActiveRPC(rpcCheckpointWorkload, cancel) defer s.clearActiveRPC() + actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + + // A checkpoint that already completed is replayed from its marker rather + // than re-run: the first one took the sandbox down, so driving runsc again + // would fail against state that no longer exists (#372). Checked before + // anything else touches the actor, including the network teardown below, + // which the completed checkpoint already did. + if rec, ok, err := checkpointmarker.Read(req.GetActorUid()); err != nil { + return nil, err + } else if ok { + slog.InfoContext(ctx, "Checkpoint already completed for this actor; replaying its result", + "actor", actorRef, + "actorUID", req.GetActorUid(), + "snapshotFiles", rec.SnapshotFiles) + return &ateompb.CheckpointWorkloadResponse{SnapshotFiles: rec.SnapshotFiles}, nil + } + if err := s.deactivateActorNetworking(ctx); err != nil { return nil, err } - actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} s.actorLogger.EmitLifecycleLog("Actor checkpointing", actorRef, req.GetActorUid(), req.GetActorTemplateNamespace(), req.GetActorTemplateName()) // Contract with atelet: @@ -715,12 +733,12 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec return nil, fmt.Errorf("no durable-dir volumes found for DATA snapshot") } if err := rcmd.cmdFsCheckpoint(ctx, "pause", checkpointPath, ddv); err != nil { - return nil, fmt.Errorf("while fscheckpointing durable-dir %q: %w", ddv[0], err) + return nil, classifyCheckpointFailure(ctx, rcmd, fmt.Errorf("while fscheckpointing durable-dir %q: %w", ddv[0], err)) } case ateompb.SnapshotScope_SNAPSHOT_SCOPE_FULL: // Checkpoint pause container (root of the sandbox) if err := rcmd.cmdCheckpoint(ctx, "pause", checkpointPath); err != nil { - return nil, fmt.Errorf("while checkpointing pause: %w", err) + return nil, classifyCheckpointFailure(ctx, rcmd, fmt.Errorf("while checkpointing pause: %w", err)) } default: return nil, fmt.Errorf("unsupported snapshot scope: %v", req.GetScope()) @@ -771,14 +789,45 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec return nil, fmt.Errorf("while listing checkpoint files: %w", err) } + // Record the result before answering, so a caller that never sees this + // response can ask again and be told the same thing. Written last: from + // here on the checkpoint is a fact on disk, whatever happens to the reply. + if err := checkpointmarker.Write(req.GetActorUid(), snapshotFiles); err != nil { + return nil, err + } + s.actorLogger.EmitLifecycleLog("Actor checkpointed", actorRef, req.GetActorUid(), req.GetActorTemplateNamespace(), req.GetActorTemplateName()) s.activeSession = nil return &ateompb.CheckpointWorkloadResponse{SnapshotFiles: snapshotFiles}, nil } +// classifyCheckpointFailure decides whether a failed checkpoint left the actor +// recoverable. A checkpoint command can fail with the sandbox still up (a +// transient runsc error, worth retrying) or with the sandbox already gone — +// the shape a replayed checkpoint takes when the first one destroyed the +// sandbox but crashed before its marker landed, which no retry can ever +// satisfy. Probing the pause container tells the two apart, so the control +// plane sees "this actor's state is unrecoverable" instead of an opaque +// `runsc` exit status. +// +// The probe runs only on the failure path: the happy path must not pay for an +// extra runsc invocation. +func classifyCheckpointFailure(ctx context.Context, rcmd *runsc, err error) error { + stateErr := rcmd.cmdState(ctx, "pause") + if stateErr == nil { + return err + } + slog.WarnContext(ctx, "Checkpoint failed and the sandbox is gone; the actor's state is unrecoverable", + "actorUID", rcmd.actorUID, "stateErr", stateErr, "err", err) + return ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonInvalidCheckpointResult, ateerrors.ActorCrashedMetadata(), + fmt.Errorf("%w: checkpoint failed and no sandbox remains to retry against: %w", ateerrors.ReasonInvalidCheckpointResult, err)) +} + // listSnapshotFiles returns the (relative) names of regular files directly under -// dir, which atelet ships to object storage as the snapshot. +// dir, which atelet ships to object storage as the snapshot. ateom's own +// completion marker shares the directory but is bookkeeping, not snapshot +// content, so it never joins the set. func listSnapshotFiles(dir string) ([]string, error) { entries, err := os.ReadDir(dir) if err != nil { @@ -786,7 +835,7 @@ func listSnapshotFiles(dir string) ([]string, error) { } var files []string for _, e := range entries { - if e.Type().IsRegular() { + if e.Type().IsRegular() && e.Name() != ateompath.CheckpointDoneFileName { files = append(files, e.Name()) } } diff --git a/cmd/ateom-microvm/checkpoint.go b/cmd/ateom-microvm/checkpoint.go index 78244df21..34c57cdc4 100644 --- a/cmd/ateom-microvm/checkpoint.go +++ b/cmd/ateom-microvm/checkpoint.go @@ -18,6 +18,7 @@ package main import ( "context" + "errors" "fmt" "log/slog" "os" @@ -29,7 +30,9 @@ import ( "github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/ch" "github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/kata" + "github.com/agent-substrate/substrate/internal/ateerrors" "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/checkpointmarker" "github.com/agent-substrate/substrate/internal/imagecache" "github.com/agent-substrate/substrate/internal/proto/ateompb" "github.com/agent-substrate/substrate/internal/resources" @@ -61,15 +64,29 @@ import ( func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.CheckpointWorkloadRequest) (*ateompb.CheckpointWorkloadResponse, error) { s.lock.Lock() defer s.lock.Unlock() - if err := s.deactivateActorNetworking(ctx); err != nil { - return nil, err - } actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} actorUID := req.GetActorUid() templateNS := req.GetActorTemplateNamespace() templateName := req.GetActorTemplateName() + // A checkpoint that already completed is replayed from its marker rather + // than re-run: the first one tore the guest down, so there is nothing left + // to pause and snapshot (#372). Checked before anything else touches the + // actor, including the network teardown and the checkpoint-dir wipe below, + // which would destroy the very evidence this reads. + if rec, ok, err := checkpointmarker.Read(actorUID); err != nil { + return nil, err + } else if ok { + slog.InfoContext(ctx, "Checkpoint already completed for this actor; replaying its result", + slog.String("id", actorUID), slog.Any("snapshot_files", rec.SnapshotFiles)) + return &ateompb.CheckpointWorkloadResponse{SnapshotFiles: rec.SnapshotFiles}, nil + } + + if err := s.deactivateActorNetworking(ctx); err != nil { + return nil, err + } + s.actorLogger.EmitLifecycleLog("Actor checkpointing", actorRef, actorUID, templateNS, templateName) // Check what the request asks for BEFORE touching the guest: these are @@ -102,6 +119,18 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec } client := ch.NewClient(chSocket) if _, err := client.WaitReady(ctx, 10*time.Second); err != nil { + // WaitReady also fails on a VMM that is merely slow, which is worth + // retrying, so only the unambiguous case is called unrecoverable: no + // api-socket at all means no VMM to snapshot. Together with the absent + // marker above, that says the actor's state is gone rather than + // pending — the shape a replayed checkpoint takes when the first one + // tore the guest down but did not live to record it. Saying so with + // the crash directive stops the control plane retrying a call that can + // never succeed. + if _, statErr := os.Stat(chSocket); errors.Is(statErr, os.ErrNotExist) { + return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonInvalidCheckpointResult, ateerrors.ActorCrashedMetadata(), + fmt.Errorf("%w: no guest remains to checkpoint: api-socket %q is gone: %w", ateerrors.ReasonInvalidCheckpointResult, chSocket, err)) + } return nil, fmt.Errorf("while waiting for CH api-socket: %w", err) } @@ -151,6 +180,13 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec return nil, fmt.Errorf("while listing snapshot files: %w", err) } + // Record the result before the teardown below and before answering, so a + // caller that never sees this response can ask again and be told the same + // thing. From here on the checkpoint is a fact on disk. + if err := checkpointmarker.Write(actorUID, snapshotFiles); err != nil { + return nil, err + } + // Tear down: the actor returns to "available". Best-effort; the snapshot is // already on disk for atelet to ship. tTeardown := time.Now() @@ -252,7 +288,9 @@ func listFiles(dir string) ([]string, error) { } var files []string for _, e := range entries { - if e.Type().IsRegular() { + // ateom's own completion marker shares the directory but is + // bookkeeping, not snapshot content, so it never joins the set. + if e.Type().IsRegular() && e.Name() != ateompath.CheckpointDoneFileName { files = append(files, e.Name()) } } diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index 5c47c2d49..0001e1382 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -152,6 +152,29 @@ func CheckpointStateDir(actorUID string) string { ) } +// CheckpointDoneFileName is the completion marker ateom writes into +// CheckpointStateDir once a checkpoint's files are all on disk, holding the +// same file list the RPC reports. Its presence is what lets a repeated +// CheckpointWorkload replay that result instead of driving the sandbox +// runtime a second time — the sandbox is gone after the first checkpoint, so +// the second attempt would fail against state that no longer exists. +// +// It is named here rather than in ateom because atelet reads the same +// directory, and both must agree the marker is not one of the snapshot's own +// files. +const CheckpointDoneFileName = "checkpoint-done.json" + +// CheckpointDoneFile is CheckpointDoneFileName inside the actor's checkpoint +// directory. It lives under CheckpointStateDir, which atelet wipes in +// resetActorDirs, so the marker's lifetime is bounded by the actor's existing +// on-node state and needs no cleanup of its own. +func CheckpointDoneFile(actorUID string) string { + return filepath.Join( + CheckpointStateDir(actorUID), + CheckpointDoneFileName, + ) +} + func LocalCheckpointsDir(actorUID string) string { return filepath.Join( ActorPath(actorUID), diff --git a/internal/checkpointmarker/checkpointmarker.go b/internal/checkpointmarker/checkpointmarker.go new file mode 100644 index 000000000..407a5d80d --- /dev/null +++ b/internal/checkpointmarker/checkpointmarker.go @@ -0,0 +1,106 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package checkpointmarker reads and writes the per-actor checkpoint +// completion marker both ateom runtimes leave beside a finished checkpoint. +// +// A checkpoint is destructive: it takes the sandbox down. That makes it the +// one workload operation whose response cannot simply be re-derived by trying +// again — a caller that loses the response (an atelet restart, a deadline +// exceeded mid-call) has no way to tell "never started" from "finished, and +// the answer went missing". Replaying it drove the sandbox runtime against +// state the first attempt had already destroyed. +// +// The marker is what makes the second attempt answerable: ateom writes it once +// the snapshot files are all on disk, recording exactly the file list it is +// about to report, and consults it before touching the runtime. Writes are +// atomic, so a marker that exists is always complete. +package checkpointmarker + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + + "github.com/agent-substrate/substrate/internal/ateompath" +) + +// Record is the marker's content: the snapshot files ateom wrote, in the same +// order it reported them, so a replayed response is identical to the original. +type Record struct { + SnapshotFiles []string `json:"snapshotFiles"` +} + +// Write records the completed checkpoint for actorUID. It writes atomically +// (temp file plus rename) so a crash mid-write leaves no marker rather than a +// truncated one that would be read as a complete checkpoint. +func Write(actorUID string, snapshotFiles []string) error { + data, err := json.Marshal(&Record{SnapshotFiles: snapshotFiles}) + if err != nil { + return fmt.Errorf("while marshaling checkpoint marker: %w", err) + } + + path := ateompath.CheckpointDoneFile(actorUID) + tmp, err := os.CreateTemp(filepath.Dir(path), "."+ateompath.CheckpointDoneFileName+".tmp-") + if err != nil { + return fmt.Errorf("while creating checkpoint marker temp file: %w", err) + } + tmpName := tmp.Name() + defer func() { + tmp.Close() + os.Remove(tmpName) // no-op once the rename below succeeds + }() + + if _, err := tmp.Write(data); err != nil { + return fmt.Errorf("while writing checkpoint marker: %w", err) + } + // Flush the bytes before the rename publishes the name: a rename over + // unsynced content can survive a node crash as an empty file. + if err := tmp.Sync(); err != nil { + return fmt.Errorf("while syncing checkpoint marker: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("while closing checkpoint marker: %w", err) + } + if err := os.Rename(tmpName, path); err != nil { + return fmt.Errorf("while renaming checkpoint marker into place: %w", err) + } + return nil +} + +// Read returns the marker recorded for actorUID. ok is false when no +// checkpoint has completed for this actor, which is the ordinary case on a +// first attempt and is not an error. +func Read(actorUID string) (_ *Record, ok bool, _ error) { + data, err := os.ReadFile(ateompath.CheckpointDoneFile(actorUID)) + if errors.Is(err, os.ErrNotExist) { + return nil, false, nil + } + if err != nil { + return nil, false, fmt.Errorf("while reading checkpoint marker: %w", err) + } + rec := &Record{} + if err := json.Unmarshal(data, rec); err != nil { + return nil, false, fmt.Errorf("while parsing checkpoint marker: %w", err) + } + // A marker naming no files cannot stand in for a checkpoint result: atelet + // rejects an empty file set as DataLoss anyway, and treating it as a + // completed checkpoint would silently commit an empty snapshot. + if len(rec.SnapshotFiles) == 0 { + return nil, false, fmt.Errorf("checkpoint marker for actor %q records no snapshot files", actorUID) + } + return rec, true, nil +} diff --git a/internal/checkpointmarker/checkpointmarker_test.go b/internal/checkpointmarker/checkpointmarker_test.go new file mode 100644 index 000000000..690afec3d --- /dev/null +++ b/internal/checkpointmarker/checkpointmarker_test.go @@ -0,0 +1,125 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package checkpointmarker + +import ( + "os" + "path/filepath" + "slices" + "testing" + + "github.com/agent-substrate/substrate/internal/ateompath" +) + +// useTempActorsDir points the shared actor-state root at a temp directory for +// the duration of the test, and creates the actor's checkpoint dir (ateom +// makes it before checkpointing). +func useTempActorsDir(t *testing.T, actorUID string) { + t.Helper() + orig := ateompath.ActorsDir + t.Cleanup(func() { ateompath.ActorsDir = orig }) + ateompath.ActorsDir = t.TempDir() + + if err := os.MkdirAll(ateompath.CheckpointStateDir(actorUID), 0o700); err != nil { + t.Fatalf("creating checkpoint dir: %v", err) + } +} + +func TestWriteThenRead(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + want := []string{"checkpoint.img", "pages.img", "pages_meta.img"} + if err := Write(actorUID, want); err != nil { + t.Fatalf("Write: %v", err) + } + + rec, ok, err := Read(actorUID) + if err != nil { + t.Fatalf("Read: %v", err) + } + if !ok { + t.Fatal("Read reported no marker after Write") + } + if !slices.Equal(rec.SnapshotFiles, want) { + t.Errorf("SnapshotFiles = %v, want %v", rec.SnapshotFiles, want) + } +} + +func TestWriteLeavesNoTempFiles(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + if err := Write(actorUID, []string{"checkpoint.img"}); err != nil { + t.Fatalf("Write: %v", err) + } + + // The atomic write renames a temp file into place; anything left beside the + // marker would be shipped as snapshot content by a caller listing the dir. + entries, err := os.ReadDir(ateompath.CheckpointStateDir(actorUID)) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + if len(entries) != 1 || entries[0].Name() != ateompath.CheckpointDoneFileName { + var got []string + for _, e := range entries { + got = append(got, e.Name()) + } + t.Errorf("checkpoint dir contents = %v, want only %q", got, ateompath.CheckpointDoneFileName) + } +} + +func TestReadNoMarker(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + // The ordinary first-attempt case: no marker is not an error, or every + // checkpoint would fail before it started. + rec, ok, err := Read(actorUID) + if err != nil { + t.Fatalf("Read: %v", err) + } + if ok || rec != nil { + t.Errorf("Read = (%v, %v), want (nil, false)", rec, ok) + } +} + +func TestReadRejectsUnusableMarker(t *testing.T) { + tests := []struct { + name string + content string + }{ + {"corrupt", "{not json"}, + // A marker naming no files cannot stand in for a checkpoint result: + // replaying it would commit an empty snapshot as though it held the + // actor's state. + {"no snapshot files", `{"snapshotFiles":[]}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + path := filepath.Join(ateompath.CheckpointStateDir(actorUID), ateompath.CheckpointDoneFileName) + if err := os.WriteFile(path, []byte(tt.content), 0o600); err != nil { + t.Fatalf("WriteFile: %v", err) + } + + if _, ok, err := Read(actorUID); err == nil { + t.Errorf("Read succeeded (ok=%v), want an error", ok) + } + }) + } +}