atelet checkpoint idempotency

cmd/atelet/main.go

@@ -508,6 +508,31 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
 	actorUID := req.GetActorUid()
 	actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()}
 
+	// A checkpoint whose snapshot is already at its destination is done, and
+	// re-running it would drive a sandbox the first attempt destroyed (#372).
+	// The control plane mints the destination once per suspend/pause and
+	// re-sends it on every re-entry, so a manifest there names THIS checkpoint
+	// and no other.
+	//
+	// Checked before the metrics defer below is installed: a replay that does
+	// no work is not a checkpoint, and recording it as one would report a
+	// near-zero duration against snapshots that take seconds to write.
+	// Checked before pruneLocalCheckpoints too, which would otherwise delete
+	// the local snapshot that proves the earlier success.
+	//
+	// Costs one small object read per external checkpoint. That is paid before
+	// the guest is paused, against an operation that goes on to move
+	// gigabytes.
+	committed, err := s.checkpointAlreadyCommitted(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	if committed {
+		slog.InfoContext(ctx, "Checkpoint already committed to its destination; nothing to do",
+			slog.Any("actor", actorRef), slog.String("actor_uid", actorUID))
+		return &ateletpb.CheckpointResponse{}, nil
+	}
+
 	// Per-phase timing, recorded on the way out so a failed checkpoint still
 	// reports the phases it completed. Phases left at zero never ran.
 	tStart := time.Now()
@@ -638,19 +663,89 @@ 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 {
 		return fmt.Errorf("while creating local checkpoint directory: %w", err)
 	}
 
 	// Move exactly the files ateom reported.
+	//
+	// A file already at the destination with nothing left at the source was
+	// moved by an earlier attempt at this same checkpoint: the rename is not
+	// repeatable, so re-entry has to recognize its own work rather than fail on
+	// the missing source. Only the manifest below commits the snapshot, so a
+	// half-moved set is exactly what an interrupted attempt leaves behind.
 	for _, fileName := range rec.SnapshotFiles {
 		src := filepath.Join(checkpointDir, fileName)
 		dst := filepath.Join(localCheckpointPath, fileName)
 		recordSnapshotSize(ctx, fileName, src, req.GetActorTemplateNamespace(), req.GetActorTemplateName())
 
-		if err := os.Rename(src, dst); err != nil {
+		err := os.Rename(src, dst)
+		if errors.Is(err, os.ErrNotExist) {
+			if _, statErr := os.Stat(dst); statErr == nil {
+				continue
+			}
+			// Gone from both sides: the snapshot cannot be assembled.
+			return wrapFileSystemErr(fmt.Sprintf("snapshot file %q is missing from both %s and %s", fileName, checkpointDir, localCheckpointPath), err)
+		}
+		if err != nil {
 			return fmt.Errorf("failed to move %s to %s: %w", src, dst, err)
 		}
 	}
@@ -767,28 +862,23 @@ 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
 		// and pruned it: the remote manifest is uploaded last, so its presence
 		// 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)

cmd/atelet/main_test.go

@@ -1253,11 +1253,17 @@ type recordingObjectStorage struct {
 	mu      sync.Mutex
 	objects map[string][]byte
 	putErr  error
+	// getErr stands in for a storage backend that is unreachable rather than
+	// empty — an error a caller must not read as "the object is not there".
+	getErr error
 }
 
 func (r *recordingObjectStorage) GetObject(_ context.Context, bucket, object string) (io.ReadCloser, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
+	if r.getErr != nil {
+		return nil, r.getErr
+	}
 	b, ok := r.objects[bucket+"/"+object]
 	if !ok {
 		return nil, fmt.Errorf("%w: Bucket:%q, Object:%q", ateerrors.ReasonFailedGetExternalObject, bucket, object)
@@ -1580,3 +1586,179 @@ func TestValidateUploadPausedCheckpointRequest(t *testing.T) {
 		})
 	}
 }
+
+// useTempActorsDir points the shared actor-state root at a temp directory, so
+// tests touching an actor's on-node paths (local snapshots, checkpoint state)
+// stay off the node's real /var/lib tree.
+func useTempActorsDir(t *testing.T) {
+	t.Helper()
+	orig := ateompath.ActorsDir
+	t.Cleanup(func() { ateompath.ActorsDir = orig })
+	ateompath.ActorsDir = t.TempDir()
+}
+
+func TestCheckpointAlreadyCommitted(t *testing.T) {
+	ctx := context.Background()
+	const manifestKey = "bucket/root/snapshots/ate-demo/counter-1-snap/manifest.json"
+
+	t.Run("external with an uploaded manifest", func(t *testing.T) {
+		s := &AteomHerder{gcsClient: &recordingObjectStorage{
+			objects: map[string][]byte{manifestKey: []byte(`{"pauseImage":"pause:v1"}`)},
+		}}
+
+		got, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest())
+		if err != nil {
+			t.Fatalf("checkpointAlreadyCommitted: %v", err)
+		}
+		if !got {
+			t.Error("committed = false, want true: the manifest is the commit marker")
+		}
+	})
+
+	t.Run("external with no manifest", func(t *testing.T) {
+		s := &AteomHerder{gcsClient: &recordingObjectStorage{}}
+
+		got, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest())
+		if err != nil {
+			t.Fatalf("checkpointAlreadyCommitted: %v", err)
+		}
+		if got {
+			t.Error("committed = true, want false")
+		}
+	})
+
+	t.Run("external probe failure is not read as uncommitted", func(t *testing.T) {
+		// Reading an unreachable bucket as "not committed" would send a
+		// destructive checkpoint down the re-run path on the strength of a
+		// failed lookup.
+		s := &AteomHerder{gcsClient: &recordingObjectStorage{getErr: errors.New("bucket unreachable")}}
+
+		if _, err := s.checkpointAlreadyCommitted(ctx, validCheckpointRequest()); err == nil {
+			t.Fatal("checkpointAlreadyCommitted succeeded, want the probe error surfaced")
+		}
+	})
+
+	t.Run("local with a written manifest", func(t *testing.T) {
+		useTempActorsDir(t)
+		req := validCheckpointRequest()
+		req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
+		req.Config = &ateletpb.CheckpointRequest_LocalConfig{
+			LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"},
+		}
+		writeLocalSnapshot(t, ateompath.LocalSnapshotDir(req.GetActorUid(), "pause-snap-1"),
+			sandboxAssetsRecord{SandboxClass: "gvisor", PauseImage: testPauseImage, SnapshotFiles: []string{"checkpoint.img"}},
+			map[string]string{"checkpoint.img": "img"})
+
+		got, err := (&AteomHerder{}).checkpointAlreadyCommitted(ctx, req)
+		if err != nil {
+			t.Fatalf("checkpointAlreadyCommitted: %v", err)
+		}
+		if !got {
+			t.Error("committed = false, want true")
+		}
+	})
+
+	t.Run("local with no snapshot dir", func(t *testing.T) {
+		useTempActorsDir(t)
+		req := validCheckpointRequest()
+		req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
+		req.Config = &ateletpb.CheckpointRequest_LocalConfig{
+			LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"},
+		}
+
+		got, err := (&AteomHerder{}).checkpointAlreadyCommitted(ctx, req)
+		if err != nil {
+			t.Fatalf("checkpointAlreadyCommitted: %v", err)
+		}
+		if got {
+			t.Error("committed = true, want false")
+		}
+	})
+}
+
+func TestCheckpointFastForwardsWhenAlreadyCommitted(t *testing.T) {
+	// No sandbox record on disk and no ateom dialer: every step after the
+	// commit probe would fail, so a success here can only come from the
+	// fast-forward.
+	useTempActorsDir(t)
+	s := &AteomHerder{gcsClient: &recordingObjectStorage{
+		objects: map[string][]byte{
+			"bucket/root/snapshots/ate-demo/counter-1-snap/manifest.json": []byte(`{"pauseImage":"pause:v1"}`),
+		},
+	}}
+
+	resp, err := s.Checkpoint(context.Background(), validCheckpointRequest())
+	if err != nil {
+		t.Fatalf("Checkpoint: %v", err)
+	}
+	if resp == nil {
+		t.Fatal("Checkpoint returned a nil response")
+	}
+}
+
+func TestMoveLocalCheckpointResumesPartialMove(t *testing.T) {
+	ctx := context.Background()
+	useTempActorsDir(t)
+
+	req := validCheckpointRequest()
+	req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
+	req.Config = &ateletpb.CheckpointRequest_LocalConfig{
+		LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"},
+	}
+	rec := &sandboxAssetsRecord{
+		SandboxClass:  "gvisor",
+		PauseImage:    testPauseImage,
+		SnapshotFiles: []string{"checkpoint.img", "pages.img"},
+	}
+
+	// The state an interrupted move leaves: one file already renamed into the
+	// snapshot dir, the other still in the checkpoint dir, no manifest.
+	checkpointDir := ateompath.CheckpointStateDir(req.GetActorUid())
+	dstDir := ateompath.LocalSnapshotDir(req.GetActorUid(), "pause-snap-1")
+	for dir, files := range map[string]map[string]string{
+		checkpointDir: {"pages.img": "pages"},
+		dstDir:        {"checkpoint.img": "img"},
+	} {
+		if err := os.MkdirAll(dir, 0o700); err != nil {
+			t.Fatalf("creating %s: %v", dir, err)
+		}
+		for name, body := range files {
+			if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil {
+				t.Fatalf("writing %s: %v", name, err)
+			}
+		}
+	}
+
+	if err := (&AteomHerder{}).moveLocalCheckpoint(ctx, req, checkpointDir, rec); err != nil {
+		t.Fatalf("moveLocalCheckpoint: %v", err)
+	}
+
+	for _, name := range append(rec.SnapshotFiles, sandboxManifestName) {
+		if _, err := os.Stat(filepath.Join(dstDir, name)); err != nil {
+			t.Errorf("%s missing from the snapshot dir: %v", name, err)
+		}
+	}
+}
+
+func TestMoveLocalCheckpointFailsWhenFileGoneFromBothSides(t *testing.T) {
+	useTempActorsDir(t)
+
+	req := validCheckpointRequest()
+	req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
+	req.Config = &ateletpb.CheckpointRequest_LocalConfig{
+		LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause-snap-1"},
+	}
+	checkpointDir := ateompath.CheckpointStateDir(req.GetActorUid())
+	if err := os.MkdirAll(checkpointDir, 0o700); err != nil {
+		t.Fatalf("creating checkpoint dir: %v", err)
+	}
+
+	rec := &sandboxAssetsRecord{SandboxClass: "gvisor", PauseImage: testPauseImage, SnapshotFiles: []string{"checkpoint.img"}}
+	err := (&AteomHerder{}).moveLocalCheckpoint(context.Background(), req, checkpointDir, rec)
+	if err == nil {
+		t.Fatal("moveLocalCheckpoint succeeded, want a failure: the snapshot cannot be assembled")
+	}
+	if !errors.Is(err, ateerrors.ReasonTerminalFileSystemError) {
+		t.Errorf("err = %v, want it tagged %v", err, ateerrors.ReasonTerminalFileSystemError)
+	}
+}