From 2020ee2f6ea0c19ddc6393b877abb37ce6a72bc3 Mon Sep 17 00:00:00 2001 From: Troy Chiu Date: Mon, 17 Aug 2026 09:31:25 -0700 Subject: [PATCH 1/5] checkpointmarker: record a completed checkpoint on disk --- internal/ateompath/ateompath.go | 23 ++ internal/checkpointmarker/checkpointmarker.go | 190 ++++++++++++++++ .../checkpointmarker/checkpointmarker_test.go | 212 ++++++++++++++++++ 3 files changed, 425 insertions(+) create mode 100644 internal/checkpointmarker/checkpointmarker.go create mode 100644 internal/checkpointmarker/checkpointmarker_test.go diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index 5c47c2d497..0001e13822 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 0000000000..a02b9c435c --- /dev/null +++ b/internal/checkpointmarker/checkpointmarker.go @@ -0,0 +1,190 @@ +// 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" + "log/slog" + "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, +// and the scope they were written for, which says which checkpoint they belong +// to. +type Record struct { + SnapshotFiles []string `json:"snapshotFiles"` + // Scope is the CheckpointWorkloadRequest's scope, stringified. A marker + // records one particular checkpoint, not merely that the actor has had + // one, so a request asking for different content is not answerable from it + // — see Read. + Scope string `json:"scope"` +} + +// 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, scope string, snapshotFiles []string) error { + // A marker naming no files can never stand in for a checkpoint result: + // atelet rejects an empty file set as DataLoss anyway, and one written here + // would be read back as unusable on every later attempt. Refuse it at the + // source rather than persisting a marker that only wedges the actor. + if len(snapshotFiles) == 0 { + return fmt.Errorf("refusing to record a checkpoint marker for actor %q with no snapshot files", actorUID) + } + // Likewise a marker that does not say which checkpoint it records: Read + // can only match it against a request's scope, so an unscoped one would be + // unusable on every later attempt. + if scope == "" { + return fmt.Errorf("refusing to record a checkpoint marker for actor %q with no scope", actorUID) + } + + data, err := json.Marshal(&Record{SnapshotFiles: snapshotFiles, Scope: scope}) + 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) + } + // Sync the parent directory as well. The rename is a directory-metadata + // change, and syncing the file's contents does not commit the name that + // makes them findable: without this a node crash can lose the marker + // entirely, which is precisely the crash the marker exists to survive. + dir, err := os.Open(filepath.Dir(path)) + if err != nil { + return fmt.Errorf("while opening checkpoint marker directory to sync: %w", err) + } + defer dir.Close() + if err := dir.Sync(); err != nil { + return fmt.Errorf("while syncing checkpoint marker directory: %w", err) + } + return nil +} + +// Read returns the marker recorded for actorUID by a checkpoint of the given +// scope. ok is false when no such checkpoint has completed, which is the +// ordinary case on a first attempt and is not an error. +// +// The scope match is what makes the marker answer "did THIS checkpoint +// finish?" rather than "has this actor been checkpointed?". The two differ +// because the marker is keyed on the actor: it names a per-actor path, and it +// outlives the attempt that wrote it until resetActorDirs clears it. A DATA +// marker replayed against a FULL request would have atelet commit a manifest +// claiming a full snapshot whose file set is only the durable-dir tar, leaving +// nothing to resume the guest from. +// +// A mismatch reports "no completed checkpoint" and leaves the marker alone: +// the request falls through to the ordinary path, where the runtime finds no +// sandbox to checkpoint and says so as unrecoverable. That is the honest +// answer for a differently-scoped checkpoint of an actor whose sandbox an +// earlier one already destroyed, and it is the marker's own record of that +// earlier checkpoint, still valid for its own retries, so it is not discarded. +func Read(actorUID, scope string) (_ *Record, ok bool, _ error) { + path := ateompath.CheckpointDoneFile(actorUID) + data, err := os.ReadFile(path) + 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 { + discardUnusable(path, actorUID, fmt.Errorf("while parsing checkpoint marker: %w", err)) + return nil, false, nil + } + // 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. Write + // refuses to produce one, so reaching this means the file was damaged. + if len(rec.SnapshotFiles) == 0 { + discardUnusable(path, actorUID, errors.New("checkpoint marker records no snapshot files")) + return nil, false, nil + } + // An unscoped marker was written by an ateom from before the scope was + // recorded. It cannot be matched, so it cannot be replayed; treat it like + // any other mismatch rather than assuming it meant the scope now asked + // for. + if rec.Scope != scope { + slog.Info("Checkpoint marker records a different checkpoint; not replaying it", + slog.String("actor_uid", actorUID), slog.String("marker_scope", rec.Scope), slog.String("requested_scope", scope)) + return nil, false, nil + } + return rec, true, nil +} + +// discardUnusable removes a marker that cannot be used, so the caller can +// report "no completed checkpoint" and re-run one. +// +// A damaged marker is not evidence of anything: it can neither replay a result +// nor prove one was produced. Keeping it would wedge the actor permanently — +// nothing else ever deletes the marker, so every retry would fail here +// identically while the control plane re-drove a pause/suspend that could +// never progress. Re-running the checkpoint is the recoverable answer: if the +// sandbox is still there the checkpoint simply succeeds, and if an earlier +// attempt already tore it down, the runtime's own failure classification +// reports that as unrecoverable rather than as a retriable error. +func discardUnusable(path, actorUID string, reason error) { + slog.Warn("Discarding unusable checkpoint marker; the checkpoint will be re-attempted", + slog.String("actor_uid", actorUID), slog.String("path", path), slog.Any("err", reason)) + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + // Not fatal: a later successful checkpoint overwrites the marker by + // rename regardless. + slog.Warn("Failed to remove unusable checkpoint marker", + slog.String("actor_uid", actorUID), slog.String("path", path), slog.Any("err", err)) + } +} diff --git a/internal/checkpointmarker/checkpointmarker_test.go b/internal/checkpointmarker/checkpointmarker_test.go new file mode 100644 index 0000000000..d9734ee2f7 --- /dev/null +++ b/internal/checkpointmarker/checkpointmarker_test.go @@ -0,0 +1,212 @@ +// 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) + } +} + +// testScope stands in for a CheckpointWorkloadRequest's stringified scope. +const testScope = "SNAPSHOT_SCOPE_FULL" + +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, testScope, want); err != nil { + t.Fatalf("Write: %v", err) + } + + rec, ok, err := Read(actorUID, testScope) + 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, testScope, []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, testScope) + if err != nil { + t.Fatalf("Read: %v", err) + } + if ok || rec != nil { + t.Errorf("Read = (%v, %v), want (nil, false)", rec, ok) + } +} + +// An unusable marker is discarded and reported as "no completed checkpoint", +// so the caller re-runs the checkpoint. Returning an error instead would wedge +// the actor: nothing else deletes the marker, so every retry would fail here +// identically while the control plane re-drove a workflow that could never +// progress. +func TestReadDiscardsUnusableMarker(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) + } + + rec, ok, err := Read(actorUID, testScope) + if err != nil { + t.Fatalf("Read: %v, want no error", err) + } + if ok || rec != nil { + t.Errorf("Read = (%v, %v), want (nil, false)", rec, ok) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("marker still on disk (err=%v), want removed", err) + } + }) + } +} + +// Write refuses an empty file set rather than persisting a marker that Read +// could only ever discard. +func TestWriteRejectsEmptyFileSet(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + if err := Write(actorUID, testScope, nil); err == nil { + t.Fatal("Write succeeded, want an error") + } + path := filepath.Join(ateompath.CheckpointStateDir(actorUID), ateompath.CheckpointDoneFileName) + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("marker written (err=%v), want none", err) + } +} + +// Write refuses an unscoped marker for the same reason it refuses an empty +// file set: Read can only match a marker against the scope asked for, so one +// without a scope could never be replayed. +func TestWriteRejectsEmptyScope(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + if err := Write(actorUID, "", []string{"checkpoint.img"}); err == nil { + t.Fatal("Write succeeded, want an error") + } + path := filepath.Join(ateompath.CheckpointStateDir(actorUID), ateompath.CheckpointDoneFileName) + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("marker written (err=%v), want none", err) + } +} + +// The marker records one particular checkpoint, not the fact that the actor +// has had one. Replaying a DATA marker against a FULL request would have +// atelet commit a manifest claiming a full snapshot whose file set is only the +// durable-dir tar, leaving nothing to resume the guest from. +func TestReadRejectsMarkerFromADifferentScope(t *testing.T) { + tests := []struct { + name string + content string + }{ + {"different scope", `{"snapshotFiles":["durable-dir.tar"],"scope":"SNAPSHOT_SCOPE_DATA"}`}, + // Written by an ateom from before the scope was recorded: unmatchable, + // so not replayable either. + {"no scope", `{"snapshotFiles":["checkpoint.img"]}`}, + } + 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) + } + + rec, ok, err := Read(actorUID, testScope) + if err != nil { + t.Fatalf("Read: %v, want no error", err) + } + if ok || rec != nil { + t.Errorf("Read = (%v, %v), want (nil, false)", rec, ok) + } + // Unlike a damaged marker, a mismatched one is still a valid + // record of the checkpoint that wrote it, and its own retries + // need it. It stays until resetActorDirs clears it. + if _, err := os.Stat(path); err != nil { + t.Errorf("marker removed (err=%v), want it left in place", err) + } + }) + } +} From a0c36d77d65e364914e9c2a2adcbfbd5325a4a7e Mon Sep 17 00:00:00 2001 From: Troy Chiu Date: Mon, 17 Aug 2026 09:31:29 -0700 Subject: [PATCH 2/5] ateom: replay a completed checkpoint instead of re-running it --- cmd/ateom-gvisor/checkpoint_test.go | 175 +++++++++++++++++++++++++++ cmd/ateom-gvisor/main.go | 101 +++++++++++++++- cmd/ateom-gvisor/runsc.go | 37 ++++-- cmd/ateom-microvm/checkpoint.go | 122 ++++++++++++++++--- cmd/ateom-microvm/checkpoint_test.go | 93 ++++++++++++++ 5 files changed, 497 insertions(+), 31 deletions(-) create mode 100644 cmd/ateom-gvisor/checkpoint_test.go create mode 100644 cmd/ateom-microvm/checkpoint_test.go diff --git a/cmd/ateom-gvisor/checkpoint_test.go b/cmd/ateom-gvisor/checkpoint_test.go new file mode 100644 index 0000000000..eb9f1f36d9 --- /dev/null +++ b/cmd/ateom-gvisor/checkpoint_test.go @@ -0,0 +1,175 @@ +//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" + "io" + "os" + "path/filepath" + "slices" + "testing" + + "github.com/agent-substrate/substrate/internal/actorlog" + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/atunnel" + "github.com/agent-substrate/substrate/internal/checkpointmarker" + "github.com/agent-substrate/substrate/internal/proto/ateompb" +) + +// testScope stands in for a CheckpointWorkloadRequest's stringified scope. +var testScope = ateompb.SnapshotScope_SNAPSHOT_SCOPE_FULL.String() + +// newCheckpointTestService builds the least AteomService CheckpointWorkload can +// be driven with. Only the collaborators it reaches before touching the sandbox +// are supplied, and each is inert: the atunnel servers hold no activation, so +// Deactivate is a no-op, and the logger writes nowhere. Everything past that +// point needs a real runsc, so a checkpoint driven here always fails — which is +// what the tests below want, since they assert on what happens *before* it. +// +// The fields are supplied rather than left zero because CheckpointWorkload +// calls straight into them: s.lock on its first line, then s.atunnelIngress and +// s.actorLogger, all of which panic on a nil pointer. +func newCheckpointTestService() *AteomService { + return &AteomService{ + lock: newCancelableMutex(), + atunnelIngress: &atunnel.Server{}, + atunnelEgress: &atunnel.Egress{}, + actorLogger: actorlog.NewActorLogger(io.Discard, false), + } +} + +// 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, testScope, []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, testScope, want); err != nil { + t.Fatalf("checkpointmarker.Write: %v", err) + } + + // No sandbox and no runsc path: reaching the checkpoint itself would fail, + // so a success here can only be the replay. + s := newCheckpointTestService() + 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) + } +} + +// A marker from a differently-scoped checkpoint is not this checkpoint's +// result, so it must not be replayed as one. The request falls through to the +// ordinary path, where the absent sandbox is reported for what it is. +func TestCheckpointWorkloadDoesNotReplayADifferentScope(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + if err := checkpointmarker.Write(actorUID, ateompb.SnapshotScope_SNAPSHOT_SCOPE_DATA.String(), []string{"durable-dir.tar"}); err != nil { + t.Fatalf("checkpointmarker.Write: %v", err) + } + + s := newCheckpointTestService() + resp, err := s.CheckpointWorkload(context.Background(), &ateompb.CheckpointWorkloadRequest{ + Atespace: "ate-demo", + ActorName: "counter-1", + ActorUid: actorUID, + Scope: ateompb.SnapshotScope_SNAPSHOT_SCOPE_FULL, + }) + // Falling through to the real checkpoint is the point: with no runsc it + // fails, where a replay would have returned the DATA file list as though it + // were this checkpoint's own result. + if err == nil { + t.Fatalf("CheckpointWorkload succeeded (files=%v), want a failure rather than a replay of the DATA snapshot", resp.GetSnapshotFiles()) + } + if slices.Contains(resp.GetSnapshotFiles(), "durable-dir.tar") { + t.Errorf("SnapshotFiles = %v, want the DATA marker's files not replayed", resp.GetSnapshotFiles()) + } +} + +// The classification hangs on this predicate, and the safe direction is +// asymmetric: failing to match leaves the checkpoint error retriable, while a +// wrong match crashes the actor permanently. Only runsc's own report that the +// container is absent counts — not the many ways the probe can fail to reach +// it. +func TestSandboxNotFound(t *testing.T) { + tests := []struct { + name string + out string + want bool + }{ + {"container absent", `error: loading container: container "pause" does not exist`, true}, + {"control server unresponsive", "error: connecting to control server: connection refused", false}, + {"runsc binary missing", "fork/exec /usr/bin/runsc: no such file or directory", false}, + {"probe timed out", "signal: killed", false}, + {"no output at all", "", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := sandboxNotFound([]byte(tt.out)); got != tt.want { + t.Errorf("sandboxNotFound(%q) = %v, want %v", tt.out, got, tt.want) + } + }) + } +} diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index f65aa93564..5edc728645 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(), req.GetScope().String()); 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,87 @@ 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(), req.GetScope().String(), 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 } +// stateProbeTimeout bounds the probe below. `runsc state` talks to the +// sandbox's control server, which after a checkpoint may never answer; the +// probe needs a deadline of its own so an unresponsive sandbox cannot hold the +// classification open for as long as the caller would allow. +const stateProbeTimeout = 15 * time.Second + +// 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 verdict is asymmetric on purpose. "Retriable" is the recoverable +// mistake: a retry that finds no sandbox arrives back here and is classified +// then. "Unrecoverable" is not — it crashes the actor permanently — so it is +// returned only on positive evidence that the sandbox is gone, never on a +// probe that merely failed to reach it. +// +// 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 { + // Probe on a context of its own. The failure being classified may itself BE + // the caller's ctx expiring, and on an expired ctx the probe cannot run at + // all — reading that as "the sandbox is gone" would turn every checkpoint + // that misses its deadline into permanent data loss. + probeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), stateProbeTimeout) + defer cancel() + + out, stateErr := rcmd.cmdStateOutput(probeCtx, "pause") + if stateErr == nil { + return err + } + if !sandboxNotFound(out) { + // `runsc state` fails for plenty of reasons that say nothing about + // whether the sandbox survived: an unusable runsc path, the probe + // timing out, or a control server that has stopped answering — which is + // expected right after a checkpoint takes the sandbox root down (see + // cleanupContainersAfterCheckpoint's caller). Keep the original, + // retriable error for all of them. + slog.WarnContext(ctx, "Checkpoint failed and the sandbox state could not be determined; leaving the failure retriable", + "actorUID", rcmd.actorUID, "stateErr", stateErr, "runscOutput", string(out), "err", err) + 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)) +} + +// sandboxNotFound reports whether runsc's output says the container it was +// asked about is not there — the one `runsc state` failure that is evidence +// the sandbox is gone rather than merely unreachable. +// +// This reads runsc's message because its exit status does not distinguish the +// cases. Failing to match is the safe direction (the checkpoint error stays +// retriable), so the match stays on runsc's own phrasing rather than anything +// looser that might catch an unrelated error. +func sandboxNotFound(runscOutput []byte) bool { + return strings.Contains(strings.ToLower(string(runscOutput)), "does not exist") +} + // 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 +877,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-gvisor/runsc.go b/cmd/ateom-gvisor/runsc.go index 4c89b799fa..58aeffb271 100644 --- a/cmd/ateom-gvisor/runsc.go +++ b/cmd/ateom-gvisor/runsc.go @@ -303,19 +303,24 @@ func (r *runsc) cmdDelete(ctx context.Context, containerName string) error { return nil } -func (r *runsc) cmdState(ctx context.Context, containerName string) error { - reapLock.RLock() - defer reapLock.RUnlock() - - cmd := exec.CommandContext( - ctx, - r.path, +// stateArgs builds the argv for `runsc state `. Factored out so the +// two forms below cannot drift, and so the argument construction can be +// unit-tested without executing runsc. +func (r *runsc) stateArgs(containerName string) []string { + return []string{ "-log-format", "json", "--alsologtostderr", "-root", ateompath.RunSCStateDir(r.actorUID), "state", containerName, - ) + } +} + +func (r *runsc) cmdState(ctx context.Context, containerName string) error { + reapLock.RLock() + defer reapLock.RUnlock() + + cmd := exec.CommandContext(ctx, r.path, r.stateArgs(containerName)...) cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr if err := cmd.Run(); err != nil { @@ -324,6 +329,22 @@ func (r *runsc) cmdState(ctx context.Context, containerName string) error { return nil } +// cmdStateOutput is cmdState for callers that need to know WHY the state call +// failed, not just that it did: `runsc state` reports "this container does not +// exist" and "I could not ask" with the same exit status, so the distinction +// lives only in its output. Captured rather than streamed to the ateom's own +// stdout/stderr, so the caller decides what to log. +func (r *runsc) cmdStateOutput(ctx context.Context, containerName string) ([]byte, error) { + reapLock.RLock() + defer reapLock.RUnlock() + + out, err := exec.CommandContext(ctx, r.path, r.stateArgs(containerName)...).CombinedOutput() + if err != nil { + return out, fmt.Errorf("while running `runsc state`: %w", err) + } + return out, nil +} + // killArgs builds the argv for `runsc kill `. Factored out // so the argument construction can be unit-tested without executing runsc. func (r *runsc) killArgs(containerName, signal string) []string { diff --git a/cmd/ateom-microvm/checkpoint.go b/cmd/ateom-microvm/checkpoint.go index 78244df216..ca67baab52 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,57 @@ 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, req.GetScope().String()); 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)) + // The marker is written before the teardown below, so an attempt that + // died in between left the VMM and its virtiofsds running, the actor + // still in s.running, and the actor network still up. Replaying the + // answer without finishing that teardown would strand the guest's + // memory on this node, let GetWorkloadStats report a checkpointed actor + // as running, and leave virtiofsd serving bundle dirs that atelet wipes + // as soon as it has this response. The teardown is best-effort and + // safe to repeat, so it runs here whether or not the first attempt got + // to it. + // + // Unless the ateom has moved on. Parts of the teardown are the ateom's, + // not the actor's — the interior network, the stats attribution — so + // running it for an actor this ateom no longer holds would cut the + // network out from under whoever holds it now. A marker outlives its + // attempt until resetActorDirs clears it, and a late retry can arrive + // after the ateom has been handed to someone else. + if held := s.activeActor.Load(); held != nil && held.UID != actorUID { + slog.WarnContext(ctx, "Not running the post-checkpoint teardown: this ateom now holds a different actor", + slog.String("id", actorUID), slog.String("active_actor_uid", held.UID)) + return &ateompb.CheckpointWorkloadResponse{SnapshotFiles: rec.SnapshotFiles}, nil + } + // A nil activeActor is not the reassignment case: it means nobody is + // holding this ateom, so there is nothing to protect, and a VMM the + // first attempt left running still needs shutting down. teardownActor + // reaches it through the conventional socket path when s.running has no + // record (ateom restarted). + ra := s.running[actorUID] + s.teardownAfterCheckpoint(ctx, actorUID, ra, ch.NewClient(chSocketFor(actorUID, ra))) + 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 @@ -96,12 +141,21 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec // The actor's CH was booted by RunWorkload or relaunched by RestoreWorkload; // either way ateom owns it and tracks its api-socket. ra := s.running[actorUID] - chSocket := kata.CLHSocketPath(actorUID) - if ra != nil && ra.apiSocket != "" { - chSocket = ra.apiSocket - } + chSocket := chSocketFor(actorUID, ra) 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,8 +205,46 @@ 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, req.GetScope().String(), 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. + dTeardown := s.teardownAfterCheckpoint(ctx, actorUID, ra, client) + + s.actorLogger.EmitLifecycleLog("Actor checkpointed", actorRef, actorUID, templateNS, templateName) + slog.InfoContext(ctx, "Actor checkpointed", slog.String("id", actorUID), slog.Any("snapshot_files", snapshotFiles), + slog.String("scope", scope.String()), slog.Duration("pause", dPause), + slog.Duration("snapshot", dSnapshot), + // The durable-dir tar runs while the guest is paused, so its cost is part + // of the suspend latency and scales with the volume's contents. + slog.Duration("durable_dir", dDurable), slog.Duration("teardown", dTeardown)) + return &ateompb.CheckpointWorkloadResponse{SnapshotFiles: snapshotFiles}, nil +} + +// chSocketFor returns the actor's CH api-socket: the one ateom recorded when it +// launched the VMM, or the conventional path when ateom has no in-memory record +// of the actor (it restarted, or the actor is already torn down). +func chSocketFor(actorUID string, ra *runningActor) string { + if ra != nil && ra.apiSocket != "" { + return ra.apiSocket + } + return kata.CLHSocketPath(actorUID) +} + +// teardownAfterCheckpoint releases what a checkpointed actor still holds on +// this ateom — the CH VMM and its virtiofsds, the running-actor entry, the +// stats attribution, and the actor network — and returns how long the teardown +// itself took. +// +// Every step is best-effort (the snapshot is already on disk) and safe to +// repeat, which is what lets the replay path run it against a teardown an +// earlier attempt may have half-finished. +func (s *AteomService) teardownAfterCheckpoint(ctx context.Context, actorUID string, ra *runningActor, client *ch.Client) time.Duration { tTeardown := time.Now() s.teardownActor(ctx, actorUID, ra, client) dTeardown := time.Since(tTeardown) @@ -163,7 +255,7 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec // would let a later GetWorkloadStats report a checkpointed actor as though it // were still running. // - // Nothing above this point clears it, unlike the gVisor ateom, which clears + // Nothing before this point clears it, unlike the gVisor ateom, which clears // as soon as its checkpoint call has taken the sandbox down. Here the guest // is only paused until this teardown, so a checkpoint that failed earlier has // left it present, and reporting its usage is then the honest answer. This is @@ -175,15 +267,7 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil { slog.WarnContext(ctx, "Failed to clean up actor network after checkpoint", slog.Any("err", err)) } - - s.actorLogger.EmitLifecycleLog("Actor checkpointed", actorRef, actorUID, templateNS, templateName) - slog.InfoContext(ctx, "Actor checkpointed", slog.String("id", actorUID), slog.Any("snapshot_files", snapshotFiles), - slog.String("scope", scope.String()), slog.Duration("pause", dPause), - slog.Duration("snapshot", dSnapshot), - // The durable-dir tar runs while the guest is paused, so its cost is part - // of the suspend latency and scales with the volume's contents. - slog.Duration("durable_dir", dDurable), slog.Duration("teardown", dTeardown)) - return &ateompb.CheckpointWorkloadResponse{SnapshotFiles: snapshotFiles}, nil + return dTeardown } // snapshotVMState captures the paused guest into checkpointDir: the CH snapshot @@ -252,7 +336,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/cmd/ateom-microvm/checkpoint_test.go b/cmd/ateom-microvm/checkpoint_test.go new file mode 100644 index 0000000000..39156e6039 --- /dev/null +++ b/cmd/ateom-microvm/checkpoint_test.go @@ -0,0 +1,93 @@ +//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" + "slices" + "testing" + + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/ateomstats" + "github.com/agent-substrate/substrate/internal/checkpointmarker" + "github.com/agent-substrate/substrate/internal/proto/ateompb" + "github.com/agent-substrate/substrate/internal/resources" +) + +// 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) { + 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) + } +} + +// A marker outlives the attempt that wrote it until resetActorDirs clears it, +// so a late retry can arrive after the ateom has been handed to another actor. +// Parts of the post-checkpoint teardown are the ateom's rather than the +// actor's — the interior network, the stats attribution — so running it then +// would cut the network out from under the actor now holding the ateom. +// +// This is the one replay path drivable from `go test`: it returns before +// reaching netlink or cloud-hypervisor, which is exactly the property under +// test. +func TestCheckpointWorkloadReplaySkipsTeardownForAReassignedAteom(t *testing.T) { + const actorUID = "actor-1" + useTempActorsDir(t, actorUID) + + want := []string{"snapshot.mem", "snapshot.state"} + if err := checkpointmarker.Write(actorUID, ateompb.SnapshotScope_SNAPSHOT_SCOPE_FULL.String(), want); err != nil { + t.Fatalf("checkpointmarker.Write: %v", err) + } + + // The ateom has moved on: it now runs a different actor. + successor := &ateomstats.ActorAttribution{ + Ref: resources.ActorRef{Atespace: "ate-demo", Name: "counter-2"}, + UID: "actor-2", + } + s := &AteomService{} + s.activeActor.Store(successor) + s.guestStats.Store(&guestStatsTarget{actorUID: successor.UID}) + + 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) + } + // The recorded result is still owed to the caller: the checkpoint did + // happen, and only its teardown is someone else's business now. + if !slices.Equal(resp.GetSnapshotFiles(), want) { + t.Errorf("SnapshotFiles = %v, want %v (the recorded result, replayed verbatim)", resp.GetSnapshotFiles(), want) + } + + if got := s.activeActor.Load(); got != successor { + t.Errorf("activeActor = %v, want the successor %v left untouched", got, successor) + } + if got := s.guestStats.Load(); got == nil || got.actorUID != successor.UID { + t.Errorf("guestStats = %v, want the successor's target left untouched", got) + } +} From e0a9476f7789807b4b154dd5d9a7bbafb8ea13bb Mon Sep 17 00:00:00 2001 From: Troy Chiu Date: Mon, 17 Aug 2026 09:32:04 -0700 Subject: [PATCH 3/5] atelet: fast-forward a Checkpoint already at its destination --- cmd/atelet/local_checkpoints.go | 21 ++- cmd/atelet/local_checkpoints_test.go | 22 ++- cmd/atelet/main.go | 188 ++++++++++++++++---- cmd/atelet/main_test.go | 257 +++++++++++++++++++++++++++ cmd/atelet/metrics.go | 27 +++ cmd/atelet/metrics_test.go | 38 ++++ 6 files changed, 517 insertions(+), 36 deletions(-) diff --git a/cmd/atelet/local_checkpoints.go b/cmd/atelet/local_checkpoints.go index b32a4bf150..75debc75dd 100644 --- a/cmd/atelet/local_checkpoints.go +++ b/cmd/atelet/local_checkpoints.go @@ -23,13 +23,22 @@ import ( "github.com/agent-substrate/substrate/internal/ateompath" ) -// pruneLocalCheckpoints removes every local snapshot of the actor. +// pruneLocalCheckpoints removes the actor's local snapshots, except the one +// named by keep (pass "" to remove them all). +// +// keep is the destination of a checkpoint currently being written. An earlier +// attempt at that same checkpoint may already have moved files into it, and +// those files are the only copy: the rename took them out of the checkpoint +// dir, and the snapshot is not committed until its manifest lands, so nothing +// would re-create them. Pruning the destination would therefore destroy a +// half-moved snapshot that the move is about to finish. +// // Best-effort: failures are logged, never fatal. -func pruneLocalCheckpoints(ctx context.Context, actorUID string) { - pruneLocalCheckpointDir(ctx, ateompath.LocalCheckpointsDir(actorUID)) +func pruneLocalCheckpoints(ctx context.Context, actorUID, keep string) { + pruneLocalCheckpointDir(ctx, ateompath.LocalCheckpointsDir(actorUID), keep) } -func pruneLocalCheckpointDir(ctx context.Context, dir string) { +func pruneLocalCheckpointDir(ctx context.Context, dir, keep string) { entries, err := os.ReadDir(dir) if err != nil { if !os.IsNotExist(err) { @@ -38,6 +47,9 @@ func pruneLocalCheckpointDir(ctx context.Context, dir string) { return } for _, entry := range entries { + if keep != "" && entry.Name() == keep { + continue + } path := filepath.Join(dir, entry.Name()) if err := os.RemoveAll(path); err != nil { slog.WarnContext(ctx, "failed to prune local checkpoint", slog.String("path", path), slog.Any("err", err)) @@ -45,5 +57,6 @@ func pruneLocalCheckpointDir(ctx context.Context, dir string) { } slog.InfoContext(ctx, "pruned local checkpoint", slog.String("path", path)) } + // Only removes the directory when it is empty, so a kept snapshot stays. _ = os.Remove(dir) } diff --git a/cmd/atelet/local_checkpoints_test.go b/cmd/atelet/local_checkpoints_test.go index c1a6ebba3a..8be2047248 100644 --- a/cmd/atelet/local_checkpoints_test.go +++ b/cmd/atelet/local_checkpoints_test.go @@ -38,7 +38,7 @@ func TestPruneRemovesEverySnapshot(t *testing.T) { writeSnapshotDir(t, dir, "pause-2") writeSnapshotDir(t, dir, "pause-3") - pruneLocalCheckpointDir(context.Background(), dir) + pruneLocalCheckpointDir(context.Background(), dir, "") if _, err := os.Stat(dir); !os.IsNotExist(err) { t.Fatalf("dir still exists (err=%v), want removed entirely", err) @@ -46,5 +46,23 @@ func TestPruneRemovesEverySnapshot(t *testing.T) { } func TestPruneMissingDirIsNoop(t *testing.T) { - pruneLocalCheckpointDir(context.Background(), filepath.Join(t.TempDir(), "absent")) + pruneLocalCheckpointDir(context.Background(), filepath.Join(t.TempDir(), "absent"), "") +} + +// The snapshot a checkpoint is currently writing into survives the prune that +// clears its superseded predecessors: a re-entered attempt may already have +// moved files there, and they exist nowhere else. +func TestPruneKeepsNamedSnapshot(t *testing.T) { + dir := t.TempDir() + writeSnapshotDir(t, dir, "pause-1") + writeSnapshotDir(t, dir, "pause-2") + + pruneLocalCheckpointDir(context.Background(), dir, "pause-2") + + if _, err := os.Stat(filepath.Join(dir, "pause-1")); !os.IsNotExist(err) { + t.Errorf("pause-1 still exists (err=%v), want pruned", err) + } + if _, err := os.Stat(filepath.Join(dir, "pause-2", "memory.img")); err != nil { + t.Errorf("pause-2 was pruned, want kept: %v", err) + } } diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 0fcd67c029..4eee76a48d 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -508,16 +508,57 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} - // 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() - var dAssets, dAteom, dPersist time.Duration + // 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. + // The dimensions both paths below report under. Built before the + // fast-forward so a replay can be counted with the same attributes a real + // checkpoint carries; sandboxClass joins it later, once the on-node record + // has been read, and attrs() omits it while it is unknown. op := snapshotOp{ templateNamespace: req.GetActorTemplateNamespace(), templateName: req.GetActorTemplateName(), kind: checkpointSnapshotKind(req), scope: ateattr.SnapshotScopeValue(req.GetScope()), } + + committed, err := s.checkpointAlreadyCommitted(ctx, req) + if err != nil { + return nil, err + } + if committed { + slog.InfoContext(ctx, "Checkpoint already committed to its destination; finishing its teardown", + slog.Any("actor", actorRef), slog.String("actor_uid", actorUID)) + s.instruments.recordCheckpointReplayed(ctx, op) + // The snapshot is committed, but the teardown that follows it is not + // part of that commit: an attempt that wrote the manifest and then died + // left the actor's volumes still mounted and its on-node dirs still + // populated. Returning success over that would hand the workflow an + // actor whose volumes it is about to detach while they are still + // node-published. Both steps are idempotent, so re-running them here + // costs nothing when the first attempt did finish. + if err := s.finishCheckpoint(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { + return nil, err + } + 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() + var dAssets, dAteom, dPersist time.Duration defer func() { s.instruments.recordCheckpoint(ctx, op, err, phase{ateattr.SnapshotPhaseSandboxAssets, dAssets}, @@ -568,8 +609,9 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe }) dAteom = time.Since(tAteom) if err != nil { - // TODO: Ateom should classify checkpoint failures, and set "should-crash" - // in the metadata if the error is not retriable. + // ateom classifies its own checkpoint failures and tags the + // unrecoverable ones with the crash directive; the wrap below preserves + // that, since status.FromError finds the ErrorInfo through it. op.failedPhase = ateattr.SnapshotPhaseAteomCheckpoint return nil, fmt.Errorf("while calling ateom.CheckpointWorkload: %w", err) } @@ -585,11 +627,16 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe sandboxRec.ActorTemplateName = req.GetActorTemplateName() sandboxRec.Scope = ateattr.SnapshotScopeValue(req.GetScope()) - // No earlier pause snapshot can ever be restored again, so remove them - // all: the actor's current state was just captured by CheckpointWorkload, - // and the control plane tracks only a single local snapshot, which this + // No earlier pause snapshot can ever be restored again, so remove them: + // the actor's current state was just captured by CheckpointWorkload, and + // the control plane tracks only a single local snapshot, which this // checkpoint either overwrites (pause) or clears (suspend). - pruneLocalCheckpoints(ctx, actorUID) + // + // This checkpoint's own destination is spared. A re-entered attempt can + // have moved part of the snapshot there already, and those files exist + // nowhere else — deleting them here would leave the move below with a file + // missing from both sides and no way to assemble the snapshot. + pruneLocalCheckpoints(ctx, actorUID, req.GetLocalConfig().GetSnapshotName()) // Pruning stays outside the persist window: it collects superseded // snapshots on both paths, so timing it as part of an external upload would @@ -614,16 +661,31 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe } dPersist = time.Since(tPersist) - if err := s.unmountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { - return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonTerminalFileSystemError, ateerrors.ActorCrashedMetadata(), fmt.Errorf("while unmounting external volumes: %w", err)) + if err := s.finishCheckpoint(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { + return nil, err + } + + return &ateletpb.CheckpointResponse{}, nil +} + +// finishCheckpoint releases what the checkpointed actor still holds on this +// node: its external volumes, and its on-node directories. +// +// Split out from Checkpoint because it runs on two paths — after a snapshot +// this call persisted, and after one an earlier attempt persisted (the +// fast-forward above). Both steps are idempotent: unmountExternalVolumes reads +// a NotFound volume as already unmounted, and resetActorDirs is remove + +// recreate. +func (s *AteomHerder) finishCheckpoint(ctx context.Context, actorUID string, volumes []*ateletpb.Volume) error { + if err := s.unmountExternalVolumes(ctx, actorUID, volumes); err != nil { + return ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonTerminalFileSystemError, ateerrors.ActorCrashedMetadata(), fmt.Errorf("while unmounting external volumes: %w", err)) } // Note: we do not crash the actor if resetting the directory fails. if err := resetActorDirs(actorUID); err != nil { - return nil, fmt.Errorf("while resetting actor dirs: %w", err) + return fmt.Errorf("while resetting actor dirs: %w", err) } - - return &ateletpb.CheckpointResponse{}, nil + return nil } func toAteomSnapshotScope(scope ateletpb.SnapshotScope) ateompb.SnapshotScope { @@ -638,6 +700,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 +763,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) } } @@ -756,8 +888,9 @@ func (s *AteomHerder) UploadPausedCheckpoint(ctx context.Context, req *ateletpb. } // The uploaded snapshot supersedes every local pause snapshot of this - // actor; free the node's disk (best-effort, like Checkpoint). - pruneLocalCheckpoints(ctx, req.GetActorUid()) + // actor; free the node's disk (best-effort, like Checkpoint). Nothing is + // half-written here — the upload above is finished — so none are kept. + pruneLocalCheckpoints(ctx, req.GetActorUid(), "") return &ateletpb.UploadPausedCheckpointResponse{}, nil } @@ -767,11 +900,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 +907,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 1bb492db44..be954384a4 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,254 @@ 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"}`), + }, + }} + + // The state an attempt that committed its snapshot and then died leaves + // behind: the actor's on-node dirs still populated, because the teardown + // after the commit never ran. + req := validCheckpointRequest() + bundleDir := ateompath.OCIBundleDir(req.GetActorUid()) + if err := os.MkdirAll(bundleDir, 0o700); err != nil { + t.Fatalf("creating bundle dir: %v", err) + } + leftover := filepath.Join(bundleDir, "leftover") + if err := os.WriteFile(leftover, []byte("x"), 0o600); err != nil { + t.Fatalf("writing leftover: %v", err) + } + + resp, err := s.Checkpoint(context.Background(), req) + if err != nil { + t.Fatalf("Checkpoint: %v", err) + } + if resp == nil { + t.Fatal("Checkpoint returned a nil response") + } + + // Fast-forwarding past the teardown would hand the workflow an actor whose + // volumes are still mounted and whose dirs still hold the last activation. + if _, err := os.Stat(leftover); !os.IsNotExist(err) { + t.Errorf("bundle dir still populated (err=%v), want the checkpoint teardown to have reset it", err) + } +} + +// The prune that clears superseded snapshots must not take the destination of +// the checkpoint being written: an attempt that moved part of the snapshot +// there and died left the only copy of those files in that directory. This +// runs the two in the order Checkpoint runs them, which is the only order in +// which the bug appears — moveLocalCheckpoint alone resumes fine. +func TestCheckpointPruneKeepsPartiallyMovedSnapshot(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-2"}, + } + rec := &sandboxAssetsRecord{ + SandboxClass: "gvisor", + PauseImage: testPauseImage, + SnapshotFiles: []string{"checkpoint.img", "pages.img"}, + } + + // One file already renamed into this checkpoint's destination, the other + // still in the checkpoint dir, no manifest — plus a superseded snapshot + // from an earlier pause, which is what prune is here to collect. + checkpointDir := ateompath.CheckpointStateDir(req.GetActorUid()) + dstDir := ateompath.LocalSnapshotDir(req.GetActorUid(), "pause-snap-2") + staleDir := ateompath.LocalSnapshotDir(req.GetActorUid(), "pause-snap-1") + for dir, files := range map[string]map[string]string{ + checkpointDir: {"pages.img": "pages"}, + dstDir: {"checkpoint.img": "img"}, + staleDir: {"checkpoint.img": "old"}, + } { + 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) + } + } + } + + pruneLocalCheckpoints(ctx, req.GetActorUid(), req.GetLocalConfig().GetSnapshotName()) + 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) + } + } + if _, err := os.Stat(staleDir); !os.IsNotExist(err) { + t.Errorf("superseded snapshot still exists (err=%v), want pruned", err) + } +} + +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/atelet/metrics.go b/cmd/atelet/metrics.go index c4adb05d54..426c00f77f 100644 --- a/cmd/atelet/metrics.go +++ b/cmd/atelet/metrics.go @@ -30,6 +30,7 @@ import ( const ( restoreDurationMetric = "ate.actor.restore.duration" checkpointDurationMetric = "ate.actor.checkpoint.duration" + checkpointReplayedMetric = "ate.actor.checkpoint.replayed" ) // snapshotPhaseBuckets have to cover both ends of a phase breakdown: a warm OCI @@ -42,6 +43,7 @@ var snapshotPhaseBuckets = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1 type Instruments struct { restoreDuration metric.Float64Histogram checkpointDuration metric.Float64Histogram + checkpointReplayed metric.Int64Counter } func NewInstruments(meter metric.Meter) (*Instruments, error) { @@ -65,9 +67,18 @@ func NewInstruments(meter metric.Meter) (*Instruments, error) { return nil, fmt.Errorf("create %s histogram: %w", checkpointDurationMetric, err) } + checkpointReplayed, err := meter.Int64Counter( + checkpointReplayedMetric, + metric.WithDescription("Checkpoints answered from a snapshot an earlier attempt had already committed, rather than re-run. Counts lost checkpoint responses, which are otherwise invisible: the recovery is silent and records no duration."), + ) + if err != nil { + return nil, fmt.Errorf("create %s counter: %w", checkpointReplayedMetric, err) + } + return &Instruments{ restoreDuration: restoreDuration, checkpointDuration: checkpointDuration, + checkpointReplayed: checkpointReplayed, }, nil } @@ -123,6 +134,22 @@ func (i *Instruments) recordCheckpoint(ctx context.Context, op snapshotOp, err e recordPhases(ctx, i.checkpointDuration, op, err, phases) } +// recordCheckpointReplayed counts a checkpoint answered from a snapshot an +// earlier attempt had already committed. +// +// Deliberately a counter and not a phase on the histogram above: a replay does +// none of the work the phases measure, and timing it would report near-zero +// durations against snapshots that take seconds to write. Without it the +// recovery leaves no trace at all — it is the one successful Checkpoint that +// records no duration — so a node quietly replaying every checkpoint would look +// like a node taking none. +func (i *Instruments) recordCheckpointReplayed(ctx context.Context, op snapshotOp) { + if i == nil || i.checkpointReplayed == nil { + return + } + i.checkpointReplayed.Add(ctx, 1, metric.WithAttributes(op.attrs()...)) +} + // recordPhases skips zero-valued phases: those never started, because the // operation died before reaching them, and reporting them as instantaneous // would drag every percentile down. diff --git a/cmd/atelet/metrics_test.go b/cmd/atelet/metrics_test.go index 1e5e13ea39..459828d914 100644 --- a/cmd/atelet/metrics_test.go +++ b/cmd/atelet/metrics_test.go @@ -168,6 +168,44 @@ func TestCheckpointDurationShape(t *testing.T) { } } +// A replayed checkpoint records no duration — it does none of the work the +// phases measure — so the counter is the only trace it leaves. It carries the +// same dimensions a real checkpoint does, so the two are comparable. +func TestCheckpointReplayedShape(t *testing.T) { + inst, reader := newTestInstruments(t) + + inst.recordCheckpointReplayed(context.Background(), snapshotOp{ + templateNamespace: testTemplateNamespace, + templateName: testTemplateName, + kind: ateattr.SnapshotKindLocal, + scope: ateattr.SnapshotScopeFull, + }) + + m := collectHistogram(t, reader, checkpointReplayedMetric) + sum, ok := m.Data.(metricdata.Sum[int64]) + if !ok { + t.Fatalf("%s is %T, want an int64 sum", m.Name, m.Data) + } + if len(sum.DataPoints) != 1 { + t.Fatalf("datapoints = %d, want 1", len(sum.DataPoints)) + } + dp := sum.DataPoints[0] + if dp.Value != 1 { + t.Errorf("value = %d, want 1", dp.Value) + } + if v := attrString(t, dp.Attributes, ateattr.SnapshotKindKey); v != ateattr.SnapshotKindLocal { + t.Errorf("snapshot kind = %q, want %q", v, ateattr.SnapshotKindLocal) + } + if v := attrString(t, dp.Attributes, ateattr.SnapshotScopeKey); v != ateattr.SnapshotScopeFull { + t.Errorf("snapshot scope = %q, want %q", v, ateattr.SnapshotScopeFull) + } + // The replay never reads the on-node record, so the sandbox class is + // genuinely unknown and must be omitted rather than sent as "". + if _, ok := dp.Attributes.Value(ateattr.SandboxClassKey); ok { + t.Error("sandbox class present, want it omitted while unknown") + } +} + // TestRecordPhasesFailurePath is the failure-path contract: a restore that dies // in the download marks ate.failure.reason on that phase and on the total, // leaves the phases that already succeeded unlabeled so their latency stays From 6f15ec67977e3c4162ab656afd4beb3e0ea5daec Mon Sep 17 00:00:00 2001 From: Troy Chiu Date: Mon, 17 Aug 2026 09:32:04 -0700 Subject: [PATCH 4/5] atelet: serialize Checkpoint attempts per actor --- cmd/atelet/actorlocks.go | 62 ++++++++++++++++++++++++++++++++++++ cmd/atelet/main.go | 27 ++++++++++++++++ cmd/atelet/main_test.go | 69 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 158 insertions(+) create mode 100644 cmd/atelet/actorlocks.go diff --git a/cmd/atelet/actorlocks.go b/cmd/atelet/actorlocks.go new file mode 100644 index 0000000000..76418572f5 --- /dev/null +++ b/cmd/atelet/actorlocks.go @@ -0,0 +1,62 @@ +// 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 "sync" + +// actorLocks serializes node-local operations that would otherwise run against +// one actor's on-node state concurrently. +// +// Recovering a lost checkpoint response makes a re-entered attempt succeed +// rather than fail, which is the point — but it also means two attempts at one +// actor can now both reach the teardown that follows a checkpoint. That +// teardown removes the actor's checkpoint dir, which is the source the other +// attempt may still be uploading from. Recognizing a repeated operation and +// excluding a concurrent one are separate problems, and solving the first does +// not solve the second. +// +// The zero value is ready to use. Entries are dropped on release, so this +// holds one entry per in-flight operation rather than one per actor the node +// has ever seen. +type actorLocks struct { + mu sync.Mutex + held map[string]struct{} +} + +// tryLock claims actorUID for the caller, reporting whether it got it. It does +// not wait: a caller that finds the actor busy has nothing useful to do +// meanwhile, since the operation it would run is the one already in progress. +// +// release is always non-nil, so `defer release()` is safe on either outcome, +// and is idempotent. +func (l *actorLocks) tryLock(actorUID string) (release func(), ok bool) { + l.mu.Lock() + defer l.mu.Unlock() + if _, busy := l.held[actorUID]; busy { + return func() {}, false + } + if l.held == nil { + l.held = map[string]struct{}{} + } + l.held[actorUID] = struct{}{} + var once sync.Once + return func() { + once.Do(func() { + l.mu.Lock() + defer l.mu.Unlock() + delete(l.held, actorUID) + }) + }, true +} diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 4eee76a48d..4d9bbb7039 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -370,6 +370,9 @@ type AteomHerder struct { mu sync.RWMutex volumePlugins map[string]volume.VolumePluginWorkerPlane csiDriverConfigLister listersv1alpha1.CSIDriverConfigLister + + // Serializes Checkpoint per actor; see actorLocks. Zero value is usable. + actorLocks actorLocks } var _ ateletpb.AteomHerderServer = (*AteomHerder)(nil) @@ -508,6 +511,30 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + // One checkpoint per actor at a time. Two attempts can be in flight at + // once — a lease expires, a new leader retries while the original handler + // is still uploading and its connection is alive — and the recovery paths + // below are what make that dangerous rather than merely wasteful: the + // second attempt now runs to completion instead of dying at ateom, and its + // finishCheckpoint wipes the checkpoint dir the first attempt is still + // reading from. The first then fails mid-upload and crashes an actor whose + // snapshot did commit. + // + // Held for the whole call, including the fast-forward below, so the + // committed-check and the teardown it leads to cannot interleave with + // another attempt's persist. + release, locked := s.actorLocks.tryLock(actorUID) + if !locked { + // Retriable, and deliberately not a queued wait: a second attempt has + // nothing to contribute while the first is moving gigabytes, and + // holding the RPC open for that long only risks the caller timing out + // on a call that was never doing anything. By the time the control + // plane retries, the first attempt has usually committed and the + // fast-forward below answers immediately. + return nil, status.Errorf(codes.Aborted, "a checkpoint for actor %s is already in progress on this node", actorUID) + } + defer release() + // 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 diff --git a/cmd/atelet/main_test.go b/cmd/atelet/main_test.go index be954384a4..c60c8bfc0c 100644 --- a/cmd/atelet/main_test.go +++ b/cmd/atelet/main_test.go @@ -1837,3 +1837,72 @@ func TestMoveLocalCheckpointFailsWhenFileGoneFromBothSides(t *testing.T) { t.Errorf("err = %v, want it tagged %v", err, ateerrors.ReasonTerminalFileSystemError) } } + +func TestActorLocks(t *testing.T) { + var l actorLocks // zero value, as embedded in AteomHerder + + release, ok := l.tryLock("actor-a") + if !ok { + t.Fatal("tryLock on a free actor returned false") + } + if _, ok := l.tryLock("actor-a"); ok { + t.Error("tryLock on a held actor returned true") + } + // Locks are per actor: a busy actor must not block any other. + releaseB, ok := l.tryLock("actor-b") + if !ok { + t.Error("tryLock on a different actor returned false") + } + releaseB() + + release() + release() // idempotent: a second release must not free a later holder's claim + if _, ok := l.tryLock("actor-a"); !ok { + t.Error("tryLock after release returned false") + } + + // Released entries are dropped rather than accumulating one per actor seen. + l.mu.Lock() + held := len(l.held) + l.mu.Unlock() + if held != 1 { + t.Errorf("held = %d entries, want 1 (only the outstanding lock)", held) + } +} + +// Recovering a lost response made a re-entered Checkpoint succeed instead of +// dying at ateom, which means two attempts at one actor can now both reach the +// teardown that wipes the checkpoint dir the other is still uploading from. +// The second attempt has to be turned away rather than run alongside the first. +func TestCheckpointRejectsConcurrentAttemptForSameActor(t *testing.T) { + useTempActorsDir(t) + // Empty storage, so the commit probe answers "not committed" and the + // unblocked actor below goes on to fail for its own reasons rather than + // short-circuiting through the fast-forward. + s := &AteomHerder{gcsClient: &recordingObjectStorage{}} + + req := validCheckpointRequest() + release, ok := s.actorLocks.tryLock(req.GetActorUid()) + if !ok { + t.Fatal("could not take the actor lock to stand in for an in-flight checkpoint") + } + defer release() + + _, err := s.Checkpoint(context.Background(), req) + if status.Code(err) != codes.Aborted { + t.Errorf("Checkpoint code = %v (err=%v), want %v", status.Code(err), err, codes.Aborted) + } + // Aborted is retriable and must not carry the crash directive: the actor is + // fine, another attempt simply holds it. + if ateerrors.ActorCrashRequested(err) { + t.Error("the concurrent-attempt rejection asks the control plane to crash the actor") + } + + // A different actor on the same node is unaffected. It fails for its own + // reasons (no sandbox record, no dialer); it must not fail as concurrent. + other := validCheckpointRequest() + other.ActorUid = "123e4567-e89b-12d3-a456-426614174001" + if _, err := s.Checkpoint(context.Background(), other); status.Code(err) == codes.Aborted { + t.Error("a checkpoint for a different actor was rejected as concurrent") + } +} From dfdb60e364451797a1ecce730b2f7583f8b2313d Mon Sep 17 00:00:00 2001 From: Troy Chiu Date: Mon, 17 Aug 2026 10:10:33 -0700 Subject: [PATCH 5/5] fix --- cmd/atelet/actorlocks.go | 31 ++- cmd/atelet/main.go | 161 ++++++++++++---- cmd/atelet/main_test.go | 179 +++++++++++++++++- cmd/ateom-gvisor/checkpoint_test.go | 46 +++++ cmd/ateom-gvisor/main.go | 56 +++++- cmd/ateom-microvm/checkpoint.go | 40 +++- cmd/ateom-microvm/checkpoint_test.go | 44 +++++ cmd/ateom-microvm/internal/kata/kata.go | 9 + cmd/ateom-microvm/restore.go | 2 +- internal/checkpointmarker/checkpointmarker.go | 10 +- .../checkpointmarker/checkpointmarker_test.go | 7 +- 11 files changed, 530 insertions(+), 55 deletions(-) diff --git a/cmd/atelet/actorlocks.go b/cmd/atelet/actorlocks.go index 76418572f5..07bf33c5e6 100644 --- a/cmd/atelet/actorlocks.go +++ b/cmd/atelet/actorlocks.go @@ -14,7 +14,12 @@ package main -import "sync" +import ( + "sync" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) // actorLocks serializes node-local operations that would otherwise run against // one actor's on-node state concurrently. @@ -27,6 +32,13 @@ import "sync" // excluding a concurrent one are separate problems, and solving the first does // not solve the second. // +// Every RPC that clears an actor's on-node state takes it, not only Checkpoint: +// UploadPausedCheckpoint prunes every local snapshot, and Run and Restore reset +// the actor's directories. Each of those destroys what an in-flight checkpoint +// is still reading from, and the lease expiry that lets two checkpoints overlap +// lets a checkpoint overlap with any of them just as easily. A lock one caller +// can walk around is not a lock. +// // The zero value is ready to use. Entries are dropped on release, so this // holds one entry per in-flight operation rather than one per actor the node // has ever seen. @@ -60,3 +72,20 @@ func (l *actorLocks) tryLock(actorUID string) (release func(), ok bool) { }) }, true } + +// lockActorFor claims actorUID for the named operation, or refuses it. +// +// The refusal is Aborted: retriable, and carrying no crash directive, because +// nothing is wrong with the actor — another operation simply holds it. It is +// deliberately not a queued wait: the caller has nothing to contribute while +// the holder moves gigabytes, and keeping the RPC open for that long only +// risks a timeout on a call that was never doing anything. By the time the +// control plane retries, the holder has usually finished, and a re-entered +// checkpoint's fast-forward answers immediately. +func (s *AteomHerder) lockActorFor(operation, actorUID string) (release func(), _ error) { + release, ok := s.actorLocks.tryLock(actorUID) + if !ok { + return nil, status.Errorf(codes.Aborted, "cannot start %s for actor %s: another operation on this actor is already in progress on this node", operation, actorUID) + } + return release, nil +} diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 4d9bbb7039..12010736ab 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -410,6 +410,14 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp * actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + // resetActorDirs below wipes the actor's checkpoint dir, which a checkpoint + // that is still uploading is reading from. See actorLocks. + release, err := s.lockActorFor("run", actorUID) + if err != nil { + return nil, err + } + defer release() + sandboxRec, err := recordFromRequest(req.GetSandboxAssets()) if err != nil { return nil, status.Error(codes.InvalidArgument, err.Error()) @@ -511,10 +519,10 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} - // One checkpoint per actor at a time. Two attempts can be in flight at - // once — a lease expires, a new leader retries while the original handler - // is still uploading and its connection is alive — and the recovery paths - // below are what make that dangerous rather than merely wasteful: the + // One node-local operation per actor at a time. Two attempts can be in + // flight at once — a lease expires, a new leader retries while the original + // handler is still uploading and its connection is alive — and the recovery + // paths below are what make that dangerous rather than merely wasteful: the // second attempt now runs to completion instead of dying at ateom, and its // finishCheckpoint wipes the checkpoint dir the first attempt is still // reading from. The first then fails mid-upload and crashes an actor whose @@ -522,19 +530,24 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe // // Held for the whole call, including the fast-forward below, so the // committed-check and the teardown it leads to cannot interleave with - // another attempt's persist. - release, locked := s.actorLocks.tryLock(actorUID) - if !locked { - // Retriable, and deliberately not a queued wait: a second attempt has - // nothing to contribute while the first is moving gigabytes, and - // holding the RPC open for that long only risks the caller timing out - // on a call that was never doing anything. By the time the control - // plane retries, the first attempt has usually committed and the - // fast-forward below answers immediately. - return nil, status.Errorf(codes.Aborted, "a checkpoint for actor %s is already in progress on this node", actorUID) + // another operation's persist. + release, err := s.lockActorFor("checkpoint", actorUID) + if err != nil { + return nil, err } defer release() + // The dimensions both paths below report under. Built before the + // fast-forward so a replay can be counted with the same attributes a real + // checkpoint carries; sandboxClass joins it later, once the on-node record + // has been read, and attrs() omits it while it is unknown. + op := snapshotOp{ + templateNamespace: req.GetActorTemplateNamespace(), + templateName: req.GetActorTemplateName(), + kind: checkpointSnapshotKind(req), + scope: ateattr.SnapshotScopeValue(req.GetScope()), + } + // 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 @@ -550,17 +563,6 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe // 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. - // The dimensions both paths below report under. Built before the - // fast-forward so a replay can be counted with the same attributes a real - // checkpoint carries; sandboxClass joins it later, once the on-node record - // has been read, and attrs() omits it while it is unknown. - op := snapshotOp{ - templateNamespace: req.GetActorTemplateNamespace(), - templateName: req.GetActorTemplateName(), - kind: checkpointSnapshotKind(req), - scope: ateattr.SnapshotScopeValue(req.GetScope()), - } - committed, err := s.checkpointAlreadyCommitted(ctx, req) if err != nil { return nil, err @@ -688,7 +690,11 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe } dPersist = time.Since(tPersist) - if err := s.finishCheckpoint(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { + // Assigns the named return rather than a fresh err, so the metrics defer + // above sees this failure: a checkpoint that dies unmounting volumes is a + // failed checkpoint, and binding a new err here would have it recorded as + // a successful one with no error.type at all. + if err = s.finishCheckpoint(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { return nil, err } @@ -743,13 +749,18 @@ func (s *AteomHerder) checkpointAlreadyCommitted(ctx context.Context, req *atele if err != nil { return false, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidObjectURL) } - return s.snapshotManifestUploaded(ctx, uri) + manifest, uploaded, err := s.fetchUploadedSnapshotManifest(ctx, uri) + if err != nil || !uploaded { + return false, err + } + return committedManifestAnswers(ctx, manifest, req), nil case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL: path := filepath.Join(ateompath.LocalSnapshotDir(req.GetActorUid(), req.GetLocalConfig().GetSnapshotName()), sandboxManifestName) - switch _, err := os.Stat(path); { + manifest, err := os.ReadFile(path) + switch { case err == nil: - return true, nil + return committedManifestAnswers(ctx, manifest, req), nil case errors.Is(err, os.ErrNotExist): return false, nil default: @@ -764,23 +775,64 @@ func (s *AteomHerder) checkpointAlreadyCommitted(ctx context.Context, req *atele } } +// committedManifestAnswers reports whether the snapshot already at this +// checkpoint's destination is the one being asked for. +// +// The destination alone does not settle that. The control plane mints it once +// per suspend/pause and re-sends it on every re-entry, but it re-derives the +// scope from the live ActorTemplate each time, so a template edited between +// two attempts sends the same destination with a different scope. Answering +// "committed" there would report a Full checkpoint over a Data-only file set, +// and the ActorSnapshot recorded afterwards would claim guest memory the +// objects do not contain — with nothing to restore from and no error anywhere +// to say so. checkpointmarker.Read guards the same replay one layer down for +// the same reason; the guard has to be here too, because this fast-forward +// answers before ateom is ever called. +// +// A mismatch, an unparsable manifest, and one written before the scope was +// recorded all report "not committed" and fall through to the ordinary path, +// where the destroyed sandbox is reported as unrecoverable. That is the honest +// answer: this checkpoint has not been taken, and cannot be. +func committedManifestAnswers(ctx context.Context, manifest []byte, req *ateletpb.CheckpointRequest) bool { + rec, err := unmarshalSandboxRecord(manifest) + if err != nil { + slog.WarnContext(ctx, "Snapshot manifest at this checkpoint's destination cannot be parsed; treating the checkpoint as not committed", + slog.String("actor_uid", req.GetActorUid()), slog.Any("err", err)) + return false + } + if want := ateattr.SnapshotScopeValue(req.GetScope()); rec.Scope != want { + slog.WarnContext(ctx, "Snapshot at this checkpoint's destination records a different scope; not treating it as this checkpoint's result", + slog.String("actor_uid", req.GetActorUid()), slog.String("manifest_scope", rec.Scope), slog.String("requested_scope", want)) + return false + } + return true +} + // snapshotManifestUploaded reports whether the snapshot at uri has its -// manifest in object storage. A missing manifest is an answer, not a failure; +// manifest in object storage. +func (s *AteomHerder) snapshotManifestUploaded(ctx context.Context, uri resources.SnapshotURI) (bool, error) { + _, uploaded, err := s.fetchUploadedSnapshotManifest(ctx, uri) + return uploaded, err +} + +// fetchUploadedSnapshotManifest returns the snapshot manifest at uri, and +// whether it is there at all. 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) { +func (s *AteomHerder) fetchUploadedSnapshotManifest(ctx context.Context, uri resources.SnapshotURI) ([]byte, 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) + return nil, 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 { + manifest, err := ategcs.FetchFromGCS(ctx, s.gcsClient, manifestURI) + if err != nil { if errors.Is(err, ateerrors.ReasonFailedGetExternalObject) { - return false, nil + return nil, false, nil } - return false, fmt.Errorf("while probing for an already-uploaded snapshot manifest: %w", err) + return nil, false, fmt.Errorf("while probing for an already-uploaded snapshot manifest: %w", err) } - return true, nil + return manifest, true, nil } func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.CheckpointRequest, checkpointDir string, rec *sandboxAssetsRecord) error { @@ -815,11 +867,19 @@ func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.Che } // Write the self-describing snapshot manifest beside the images. + // + // Atomically, because this file is what commits the snapshot: + // checkpointAlreadyCommitted reads its presence as proof that every file it + // names is in place. os.WriteFile truncates before it writes, so a node that + // died mid-write would leave an empty manifest that the next attempt would + // fast-forward over, reporting a checkpoint whose manifest cannot be parsed + // — a failure that would only surface later, at upload or restore, as + // unrecoverable. A crash now leaves the previous state instead. manifest, err := json.Marshal(rec) if err != nil { return fmt.Errorf("while marshaling snapshot manifest: %w", err) } - if err := os.WriteFile(filepath.Join(localCheckpointPath, sandboxManifestName), manifest, 0o600); err != nil { + if err := writeFileAtomic(filepath.Join(localCheckpointPath, sandboxManifestName), manifest, 0o600); err != nil { return fmt.Errorf("while writing snapshot manifest: %w", err) } @@ -883,6 +943,15 @@ func (s *AteomHerder) UploadPausedCheckpoint(ctx context.Context, req *ateletpb. return nil, status.Error(codes.InvalidArgument, err.Error()) } + // The prune at the end of this call removes every local snapshot of the + // actor, including the destination a concurrent local checkpoint is part + // way through renaming files into. See actorLocks. + release, err := s.lockActorFor("paused-checkpoint upload", req.GetActorUid()) + if err != nil { + return nil, err + } + defer release() + tStart := time.Now() var dPersist time.Duration op := snapshotOp{ @@ -1010,6 +1079,14 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + // resetActorDirs below wipes the actor's checkpoint dir, which a checkpoint + // that is still uploading is reading from. See actorLocks. + release, err := s.lockActorFor("restore", actorUID) + if err != nil { + return nil, err + } + defer release() + // Per-step timing so we can attribute resume latency between the rustfs // download/decompress, the OCI image unpack, and ateom's own work. Logged at // the end, and recorded per phase on the way out so a failed restore still @@ -1888,9 +1965,13 @@ func validateUploadPausedCheckpointRequest(req *ateletpb.UploadPausedCheckpointR // writeFileAtomic writes data to path by writing a temp file in the same // directory, syncing, and renaming it over the target, then syncing the -// parent directory so the rename is durable. The identity directory is -// bind-mounted into actors, so the file must change atomically: a reader -// must never observe a truncated or partially written value. +// parent directory so the rename is durable, so that no reader ever observes a +// truncated or partially written value. +// +// Its callers are the files where a half-written value would be believed: the +// identity directory is bind-mounted into actors, which read it live, and the +// local snapshot manifest is what commits a checkpoint, so a truncated one +// would be read by a later attempt as a snapshot that is complete. func writeFileAtomic(path string, data []byte, perm os.FileMode) error { f, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+".tmp-*") if err != nil { diff --git a/cmd/atelet/main_test.go b/cmd/atelet/main_test.go index c60c8bfc0c..55241727b3 100644 --- a/cmd/atelet/main_test.go +++ b/cmd/atelet/main_test.go @@ -1603,7 +1603,7 @@ func TestCheckpointAlreadyCommitted(t *testing.T) { t.Run("external with an uploaded manifest", func(t *testing.T) { s := &AteomHerder{gcsClient: &recordingObjectStorage{ - objects: map[string][]byte{manifestKey: []byte(`{"pauseImage":"pause:v1"}`)}, + objects: map[string][]byte{manifestKey: []byte(`{"pauseImage":"pause:v1","scope":"full"}`)}, }} got, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest()) @@ -1615,6 +1615,57 @@ func TestCheckpointAlreadyCommitted(t *testing.T) { } }) + // The destination is minted once per operation and re-sent on every + // re-entry, but the scope is re-derived from the live ActorTemplate each + // time. A template edited between two attempts therefore aims a Full + // checkpoint at a destination holding a Data snapshot; answering + // "committed" would report guest memory that was never captured. + t.Run("external manifest recording a different scope", func(t *testing.T) { + s := &AteomHerder{gcsClient: &recordingObjectStorage{ + objects: map[string][]byte{manifestKey: []byte(`{"pauseImage":"pause:v1","scope":"data"}`)}, + }} + + req := validCheckpointRequest() // FULL + got, err := s.checkpointAlreadyCommitted(ctx, req) + if err != nil { + t.Fatalf("checkpointAlreadyCommitted: %v", err) + } + if got { + t.Errorf("committed = true, want false: the destination holds a %s snapshot, not the %s one asked for", + ateattr.SnapshotScopeData, ateattr.SnapshotScopeValue(req.GetScope())) + } + }) + + // Written before the scope was recorded in the manifest. It cannot be + // matched, so it cannot answer for this checkpoint. + t.Run("external manifest with no scope recorded", 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 = true, want false: an unscoped manifest cannot be matched to this checkpoint") + } + }) + + t.Run("external manifest that cannot be parsed", func(t *testing.T) { + s := &AteomHerder{gcsClient: &recordingObjectStorage{ + objects: map[string][]byte{manifestKey: []byte("not json")}, + }} + + got, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest()) + if err != nil { + t.Fatalf("checkpointAlreadyCommitted: %v", err) + } + if got { + t.Error("committed = true, want false: a manifest that cannot be read proves nothing") + } + }) + t.Run("external with no manifest", func(t *testing.T) { s := &AteomHerder{gcsClient: &recordingObjectStorage{}} @@ -1646,7 +1697,7 @@ func TestCheckpointAlreadyCommitted(t *testing.T) { 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"}}, + sandboxAssetsRecord{SandboxClass: "gvisor", PauseImage: testPauseImage, SnapshotFiles: []string{"checkpoint.img"}, Scope: ateattr.SnapshotScopeFull}, map[string]string{"checkpoint.img": "img"}) got, err := (&AteomHerder{}).checkpointAlreadyCommitted(ctx, req) @@ -1658,6 +1709,27 @@ func TestCheckpointAlreadyCommitted(t *testing.T) { } }) + t.Run("local snapshot recording a different scope", func(t *testing.T) { + useTempActorsDir(t) + req := validCheckpointRequest() // FULL + 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{"durable-dir.tar"}, Scope: ateattr.SnapshotScopeData}, + map[string]string{"durable-dir.tar": "tar"}) + + got, err := (&AteomHerder{}).checkpointAlreadyCommitted(ctx, req) + if err != nil { + t.Fatalf("checkpointAlreadyCommitted: %v", err) + } + if got { + t.Errorf("committed = true, want false: the destination holds a %s snapshot, not the %s one asked for", + ateattr.SnapshotScopeData, ateattr.SnapshotScopeValue(req.GetScope())) + } + }) + t.Run("local with no snapshot dir", func(t *testing.T) { useTempActorsDir(t) req := validCheckpointRequest() @@ -1683,7 +1755,7 @@ func TestCheckpointFastForwardsWhenAlreadyCommitted(t *testing.T) { useTempActorsDir(t) s := &AteomHerder{gcsClient: &recordingObjectStorage{ objects: map[string][]byte{ - "bucket/root/snapshots/ate-demo/counter-1-snap/manifest.json": []byte(`{"pauseImage":"pause:v1"}`), + "bucket/root/snapshots/ate-demo/counter-1-snap/manifest.json": []byte(`{"pauseImage":"pause:v1","scope":"full"}`), }, }} @@ -1815,6 +1887,59 @@ func TestMoveLocalCheckpointResumesPartialMove(t *testing.T) { } } +// checkpointAlreadyCommitted reads the manifest's mere presence as proof the +// snapshot is complete, so the snapshot dir must hold nothing else that could +// be mistaken for it and nothing left over from writing it: the manifest is +// renamed into place from a temp file in the same directory, and a temp file +// that outlived its write would join the snapshot's own contents. +func TestMoveLocalCheckpointLeavesOnlyTheCommittedSnapshot(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"}, + } + rec := &sandboxAssetsRecord{SandboxClass: "gvisor", PauseImage: testPauseImage, SnapshotFiles: []string{"checkpoint.img"}} + + checkpointDir := ateompath.CheckpointStateDir(req.GetActorUid()) + if err := os.MkdirAll(checkpointDir, 0o700); err != nil { + t.Fatalf("creating checkpoint dir: %v", err) + } + if err := os.WriteFile(filepath.Join(checkpointDir, "checkpoint.img"), []byte("img"), 0o600); err != nil { + t.Fatalf("writing checkpoint.img: %v", err) + } + + if err := (&AteomHerder{}).moveLocalCheckpoint(context.Background(), req, checkpointDir, rec); err != nil { + t.Fatalf("moveLocalCheckpoint: %v", err) + } + + dstDir := ateompath.LocalSnapshotDir(req.GetActorUid(), "pause-snap-1") + entries, err := os.ReadDir(dstDir) + if err != nil { + t.Fatalf("reading snapshot dir: %v", err) + } + var got []string + for _, e := range entries { + got = append(got, e.Name()) + } + slices.Sort(got) + want := []string{"checkpoint.img", sandboxManifestName} + if !slices.Equal(got, want) { + t.Errorf("snapshot dir = %v, want exactly %v", got, want) + } + + // A manifest that is present but unreadable is the state the fast-forward + // cannot detect, so it must never be committed. + manifest, err := os.ReadFile(filepath.Join(dstDir, sandboxManifestName)) + if err != nil { + t.Fatalf("reading manifest: %v", err) + } + if _, err := unmarshalSandboxRecord(manifest); err != nil { + t.Errorf("unmarshalSandboxRecord: %v, want the committed manifest to parse", err) + } +} + func TestMoveLocalCheckpointFailsWhenFileGoneFromBothSides(t *testing.T) { useTempActorsDir(t) @@ -1906,3 +2031,51 @@ func TestCheckpointRejectsConcurrentAttemptForSameActor(t *testing.T) { t.Error("a checkpoint for a different actor was rejected as concurrent") } } + +// The lock is only worth having if every RPC that clears the actor's on-node +// state takes it: UploadPausedCheckpoint prunes its local snapshots, and Run +// and Restore reset its directories, each of them destroying what an in-flight +// checkpoint is still reading from. +func TestActorLockExcludesTheOtherNodeLocalOperations(t *testing.T) { + ctx := context.Background() + const actorUID = "123e4567-e89b-12d3-a456-426614174000" + + tests := []struct { + name string + call func(*AteomHerder) error + }{ + {"Run", func(s *AteomHerder) error { + _, err := s.Run(ctx, validRunRequest()) + return err + }}, + {"Restore", func(s *AteomHerder) error { + _, err := s.Restore(ctx, validRestoreRequest()) + return err + }}, + {"UploadPausedCheckpoint", func(s *AteomHerder) error { + _, err := s.UploadPausedCheckpoint(ctx, validUploadPausedCheckpointRequest()) + return err + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + useTempActorsDir(t) + s := &AteomHerder{gcsClient: &recordingObjectStorage{}} + + release, ok := s.actorLocks.tryLock(actorUID) + if !ok { + t.Fatal("could not take the actor lock to stand in for an in-flight checkpoint") + } + defer release() + + err := tt.call(s) + if status.Code(err) != codes.Aborted { + t.Errorf("%s code = %v (err=%v), want %v", tt.name, status.Code(err), err, codes.Aborted) + } + // The actor is fine; another operation simply holds it. + if ateerrors.ActorCrashRequested(err) { + t.Errorf("%s's concurrent-operation rejection asks the control plane to crash the actor", tt.name) + } + }) + } +} diff --git a/cmd/ateom-gvisor/checkpoint_test.go b/cmd/ateom-gvisor/checkpoint_test.go index eb9f1f36d9..4fc9d925c7 100644 --- a/cmd/ateom-gvisor/checkpoint_test.go +++ b/cmd/ateom-gvisor/checkpoint_test.go @@ -18,6 +18,7 @@ package main import ( "context" + "errors" "io" "os" "path/filepath" @@ -164,6 +165,22 @@ func TestSandboxNotFound(t *testing.T) { {"runsc binary missing", "fork/exec /usr/bin/runsc: no such file or directory", false}, {"probe timed out", "signal: killed", false}, {"no output at all", "", false}, + // We capture runsc's whole --alsologtostderr stream, and gVisor logs + // "does not exist" about things it merely probed on its way to an + // unrelated failure. Only the verdict line counts: reading a log line as + // the verdict would crash an actor whose sandbox is alive. + { + "phrase in an incidental log line, verdict says otherwise", + `{"msg":"cgroup path \"/sys/fs/cgroup/runsc\" does not exist, skipping","level":"warning"} +error: connecting to control server: connection refused`, + false, + }, + { + "verdict after log noise", + `{"msg":"loading container","level":"info"} +error: loading container: container "pause" does not exist`, + true, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -173,3 +190,32 @@ func TestSandboxNotFound(t *testing.T) { }) } } + +// A retried checkpoint must not inherit the previous attempt's images: they +// would join the manifest through listSnapshotFiles and reach a restore as +// pages from a checkpoint that never completed. +func TestCheckpointWorkloadClearsStaleCheckpointFiles(t *testing.T) { + const actorUID = "actor-1" + dir := useTempActorsDir(t, actorUID) + + stale := filepath.Join(dir, "pages.img") + if err := os.WriteFile(stale, []byte("half-written"), 0o600); err != nil { + t.Fatalf("writing stale image: %v", err) + } + + // No marker and no runsc: the checkpoint itself fails, which is fine — the + // dir is cleared on the way there, before anything is written. + s := newCheckpointTestService() + if _, err := s.CheckpointWorkload(context.Background(), &ateompb.CheckpointWorkloadRequest{ + Atespace: "ate-demo", + ActorName: "counter-1", + ActorUid: actorUID, + Scope: ateompb.SnapshotScope_SNAPSHOT_SCOPE_FULL, + }); err == nil { + t.Fatal("CheckpointWorkload succeeded, want a failure with no runsc to drive") + } + + if _, err := os.Stat(stale); !errors.Is(err, os.ErrNotExist) { + t.Errorf("os.Stat(%q) = %v, want the previous attempt's image to be gone", stale, err) + } +} diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 5edc728645..bafbdba285 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -715,6 +715,23 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec } checkpointPath := ateompath.CheckpointStateDir(req.GetActorUid()) + // Start from a clean dir so runsc's image files are the only contents. A + // checkpoint that failed with the sandbox still up is retried through here, + // and its half-written images would otherwise survive: listSnapshotFiles + // reports the union of both attempts, so the manifest — and the image + // directory a later restore is handed — would carry pages from a checkpoint + // that never completed. + // + // An unmarked dir is not proof that no checkpoint completed: the marker + // write below is allowed to fail. So this can also be a complete snapshot + // whose marker never landed and whose response went missing — and it is + // deleted, deliberately. The two states are indistinguishable from here, + // and that one is already lost: atelet never received the file list, so + // nothing can name those images again. Keeping them would only trade a + // certain bug (stale pages joining the next snapshot) for a dead copy. + if err := os.RemoveAll(checkpointPath); err != nil { + return nil, fmt.Errorf("while clearing checkpoint directory: %w", err) + } if err := os.MkdirAll(checkpointPath, 0o700); err != nil { return nil, fmt.Errorf("while creating checkpoint directory: %w", err) } @@ -792,8 +809,26 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec // 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. + // + // A marker that cannot be written is logged and no more: it buys re-entry, + // it is not what makes the checkpoint valid. By this point the snapshot is + // complete on disk and the sandbox is gone, so failing here would withhold + // a file list nobody can produce again — atelet would never ship the + // snapshot, and the retry would find no sandbox and crash the actor. + // Answering leaves only the narrower risk the marker exists to cover: a + // response that goes missing. ENOSPC is the case to expect, the checkpoint + // above having just written its images to this same filesystem. + // + // The micro-VM ateom does the opposite and fails, because there the guest + // is merely paused until the teardown that follows: its checkpoint can be + // re-run in full, so refusing to answer without a marker costs nothing and + // keeps the response and the marker in step. if err := checkpointmarker.Write(req.GetActorUid(), req.GetScope().String(), snapshotFiles); err != nil { - return nil, err + slog.ErrorContext(ctx, "Failed to record the checkpoint completion marker; answering anyway, but a lost response can no longer be replayed", + "actor", actorRef, + "actorUID", req.GetActorUid(), + "snapshotFiles", snapshotFiles, + "err", err) } s.actorLogger.EmitLifecycleLog("Actor checkpointed", actorRef, req.GetActorUid(), req.GetActorTemplateNamespace(), req.GetActorTemplateName()) @@ -862,8 +897,25 @@ func classifyCheckpointFailure(ctx context.Context, rcmd *runsc, err error) erro // cases. Failing to match is the safe direction (the checkpoint error stays // retriable), so the match stays on runsc's own phrasing rather than anything // looser that might catch an unrelated error. +// +// Only the `error:` line counts. We run runsc with --alsologtostderr, so the +// captured output is runsc's whole log stream, and gVisor logs "does not +// exist" about incidental things it probes along the way (an absent cgroup +// path, a missing file). Matching the stream as a whole would let one of those +// lines crash an actor whose sandbox is alive and whose checkpoint failure was +// retriable. runsc's own verdict is the single line its fatal path writes with +// an `error: ` prefix. func sandboxNotFound(runscOutput []byte) bool { - return strings.Contains(strings.ToLower(string(runscOutput)), "does not exist") + for line := range strings.Lines(string(runscOutput)) { + msg, ok := strings.CutPrefix(strings.TrimSpace(line), "error:") + if !ok { + continue + } + if strings.Contains(strings.ToLower(msg), "does not exist") { + return true + } + } + return false } // listSnapshotFiles returns the (relative) names of regular files directly under diff --git a/cmd/ateom-microvm/checkpoint.go b/cmd/ateom-microvm/checkpoint.go index ca67baab52..9e7348ca28 100644 --- a/cmd/ateom-microvm/checkpoint.go +++ b/cmd/ateom-microvm/checkpoint.go @@ -208,6 +208,11 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec // 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. + // + // Failing here is safe, unlike in the gVisor ateom: the guest is only + // paused until the teardown below, so a retry re-runs this checkpoint from + // the top and succeeds. Refusing to answer without a marker therefore costs + // nothing and keeps the response and the marker in step. if err := checkpointmarker.Write(actorUID, req.GetScope().String(), snapshotFiles); err != nil { return nil, err } @@ -227,13 +232,42 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec } // chSocketFor returns the actor's CH api-socket: the one ateom recorded when it -// launched the VMM, or the conventional path when ateom has no in-memory record +// launched the VMM, or a conventional path when ateom has no in-memory record // of the actor (it restarted, or the actor is already torn down). +// +// Without a record there are two conventions to choose between, because +// RunWorkload and RestoreWorkload launch their VMMs on different paths, and +// nothing left on the node says which one this actor came up through. So the +// socket that exists wins. Guessing the boot path for a restored actor would +// aim a shutdown at a socket its VMM never listened on — and would have the +// caller below read "this path is absent" as "no guest remains", crashing an +// actor whose VMM is alive on the other one. func chSocketFor(actorUID string, ra *runningActor) string { + return firstExistingPath(chSocketCandidates(actorUID, ra)) +} + +// chSocketCandidates lists the api-socket paths the actor's VMM could be +// listening on, likeliest first. One when ateom knows which socket it launched +// the VMM on; otherwise both conventions, since the record is what would have +// said whether this actor was booted or restored. +func chSocketCandidates(actorUID string, ra *runningActor) []string { if ra != nil && ra.apiSocket != "" { - return ra.apiSocket + return []string{ra.apiSocket} + } + return []string{kata.CLHSocketPath(actorUID), kata.RestoredCLHSocketPath(actorUID)} +} + +// firstExistingPath returns the first candidate that is present, or the first +// candidate when none is. None being present is an answer in itself — the +// caller reads it as the guest being gone — so the likeliest path is returned +// for the error to name. +func firstExistingPath(candidates []string) string { + for _, path := range candidates { + if _, err := os.Stat(path); err == nil { + return path + } } - return kata.CLHSocketPath(actorUID) + return candidates[0] } // teardownAfterCheckpoint releases what a checkpointed actor still holds on diff --git a/cmd/ateom-microvm/checkpoint_test.go b/cmd/ateom-microvm/checkpoint_test.go index 39156e6039..f339c211aa 100644 --- a/cmd/ateom-microvm/checkpoint_test.go +++ b/cmd/ateom-microvm/checkpoint_test.go @@ -19,9 +19,11 @@ package main import ( "context" "os" + "path/filepath" "slices" "testing" + "github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/kata" "github.com/agent-substrate/substrate/internal/ateompath" "github.com/agent-substrate/substrate/internal/ateomstats" "github.com/agent-substrate/substrate/internal/checkpointmarker" @@ -91,3 +93,45 @@ func TestCheckpointWorkloadReplaySkipsTeardownForAReassignedAteom(t *testing.T) t.Errorf("guestStats = %v, want the successor's target left untouched", got) } } + +// RunWorkload and RestoreWorkload launch their VMMs on different api-socket +// paths, so with no record of which one this actor came up through, ateom +// cannot assume the boot path. Guessing wrong aims the teardown at a socket +// nothing is listening on, and has CheckpointWorkload read that absence as "no +// guest remains" and crash an actor whose VMM is alive on the other socket. +func TestCHSocketCandidates(t *testing.T) { + const actorUID = "actor-1" + + t.Run("no record covers both conventions", func(t *testing.T) { + got := chSocketCandidates(actorUID, nil) + want := []string{kata.CLHSocketPath(actorUID), kata.RestoredCLHSocketPath(actorUID)} + if !slices.Equal(got, want) { + t.Errorf("chSocketCandidates = %v, want %v", got, want) + } + }) + + t.Run("a recorded socket settles it", func(t *testing.T) { + got := chSocketCandidates(actorUID, &runningActor{apiSocket: "/run/recorded.sock"}) + if !slices.Equal(got, []string{"/run/recorded.sock"}) { + t.Errorf("chSocketCandidates = %v, want only the recorded socket", got) + } + }) +} + +func TestFirstExistingPath(t *testing.T) { + dir := t.TempDir() + missing := filepath.Join(dir, "clh-api.sock") + present := filepath.Join(dir, "clh-api-restore.sock") + if err := os.WriteFile(present, nil, 0o600); err != nil { + t.Fatalf("creating socket file: %v", err) + } + + if got := firstExistingPath([]string{missing, present}); got != present { + t.Errorf("firstExistingPath = %q, want the one that exists (%q)", got, present) + } + // None of them present is an answer too: the caller reads it as the guest + // being gone, so the likeliest path is what the error should name. + if got := firstExistingPath([]string{missing, missing + ".2"}); got != missing { + t.Errorf("firstExistingPath = %q, want the likeliest path %q", got, missing) + } +} diff --git a/cmd/ateom-microvm/internal/kata/kata.go b/cmd/ateom-microvm/internal/kata/kata.go index 0b09241ec5..4a493f465d 100644 --- a/cmd/ateom-microvm/internal/kata/kata.go +++ b/cmd/ateom-microvm/internal/kata/kata.go @@ -37,3 +37,12 @@ const vcVMDir = "/run/vc/vm" func CLHSocketPath(id string) string { return filepath.Join(vcVMDir, id, "clh-api.sock") } + +// RestoredCLHSocketPath is CLHSocketPath's counterpart for a VMM relaunched by +// a restore, which listens on a path of its own. Named here beside the boot +// path so that code reaching for an actor's api-socket without ateom's own +// record of it has both conventions in one place, rather than assuming the +// actor was booted rather than restored. +func RestoredCLHSocketPath(id string) string { + return filepath.Join(vcVMDir, id, "clh-api-restore.sock") +} diff --git a/cmd/ateom-microvm/restore.go b/cmd/ateom-microvm/restore.go index 229640e582..8f45f71c98 100644 --- a/cmd/ateom-microvm/restore.go +++ b/cmd/ateom-microvm/restore.go @@ -289,7 +289,7 @@ func (s *AteomService) restoreFullScope(ctx context.Context, p actorBootParams, // Relaunch CH and restore with the tap FDs attached (SCM_RIGHTS). CH reopens // /dev/vda (image) + each /dev/vd{b+i} (actor rootfs) from the snapshot config paths. - apiSocket := filepath.Join(kata.VMDir(actorUID), "clh-api-restore.sock") + apiSocket := kata.RestoredCLHSocketPath(actorUID) chCmd, client, err := ch.LaunchVMM(ctx, ch.LaunchVMMOptions{ Binary: rr.chBinary, APISocket: apiSocket, Stdout: slogWriter{ctx}, Stderr: slogWriter{ctx}, }) diff --git a/internal/checkpointmarker/checkpointmarker.go b/internal/checkpointmarker/checkpointmarker.go index a02b9c435c..f7fe763cde 100644 --- a/internal/checkpointmarker/checkpointmarker.go +++ b/internal/checkpointmarker/checkpointmarker.go @@ -131,8 +131,14 @@ func Write(actorUID, scope string, snapshotFiles []string) error { // the request falls through to the ordinary path, where the runtime finds no // sandbox to checkpoint and says so as unrecoverable. That is the honest // answer for a differently-scoped checkpoint of an actor whose sandbox an -// earlier one already destroyed, and it is the marker's own record of that -// earlier checkpoint, still valid for its own retries, so it is not discarded. +// earlier one already destroyed. +// +// Leaving it alone is about what Read may do, not about how long the marker +// survives: a record Read cannot use is still not Read's to delete, unlike the +// damaged one below that nobody can use. The fall-through usually removes it +// moments later anyway — both runtimes clear the checkpoint dir before taking +// a checkpoint — so this is not a promise that a mismatched marker outlives +// the call. func Read(actorUID, scope string) (_ *Record, ok bool, _ error) { path := ateompath.CheckpointDoneFile(actorUID) data, err := os.ReadFile(path) diff --git a/internal/checkpointmarker/checkpointmarker_test.go b/internal/checkpointmarker/checkpointmarker_test.go index d9734ee2f7..8e0edb3102 100644 --- a/internal/checkpointmarker/checkpointmarker_test.go +++ b/internal/checkpointmarker/checkpointmarker_test.go @@ -201,9 +201,10 @@ func TestReadRejectsMarkerFromADifferentScope(t *testing.T) { if ok || rec != nil { t.Errorf("Read = (%v, %v), want (nil, false)", rec, ok) } - // Unlike a damaged marker, a mismatched one is still a valid - // record of the checkpoint that wrote it, and its own retries - // need it. It stays until resetActorDirs clears it. + // Unlike a damaged marker, a mismatched one is a valid record of + // the checkpoint that wrote it, so it is not Read's to delete. + // (Its caller will usually clear the checkpoint dir moments + // later; that is the caller's decision, not this one's.) if _, err := os.Stat(path); err != nil { t.Errorf("marker removed (err=%v), want it left in place", err) }