From d01066f58237fb8129503992764d3f2705d4160d Mon Sep 17 00:00:00 2001 From: Anish Gangu Date: Fri, 6 Nov 2026 21:00:54 +0000 Subject: [PATCH] atelet: move snapshot-state.json back from the snapshot directory at startup This is the first of two releases that move each actor's snapshot-state.json from /var/lib/ate/actors// to /var/lib/ate/snapshots//, so its writes stop waiting on the boot disk. The version skew policy needs a node layout change to span two minor releases. This release still keeps the file in the old place. At startup, before the snapshot tracker is created, atelet moves any file the next release left in the snapshot directory back to the old place, so a rollback from that release keeps every actor's snapshots. Only one atelet runs on a node, so such a file was written before startup, and the rest of atelet keeps using one place. Each moved file is synced before the original is removed, so a crash leaves two identical copies that the next startup finishes moving. A file whose actor directory is gone is removed instead. atelet exits if a move fails, so it never loses track of an actor's snapshots. Co-authored-by: Michelle Au Signed-off-by: Anish Gangu --- cmd/atelet/internal/ateletpath/ateletpath.go | 16 +- .../internal/ateletpath/ateletpath_test.go | 10 +- cmd/atelet/internal/snapshottracker/state.go | 63 +++++++ .../internal/snapshottracker/state_test.go | 176 ++++++++++++++++++ cmd/atelet/main.go | 3 + 5 files changed, 263 insertions(+), 5 deletions(-) create mode 100644 cmd/atelet/internal/snapshottracker/state_test.go diff --git a/cmd/atelet/internal/ateletpath/ateletpath.go b/cmd/atelet/internal/ateletpath/ateletpath.go index b372ff5fb6..dcacde502d 100644 --- a/cmd/atelet/internal/ateletpath/ateletpath.go +++ b/cmd/atelet/internal/ateletpath/ateletpath.go @@ -109,10 +109,24 @@ func ActorSandboxAssetsFile(actorUID string) string { func ActorSnapshotStateFile(actorUID string) string { return filepath.Join( ActorPath(actorUID), - "snapshot-state.json", + snapshotStateFileName, ) } +// NextSnapshotStateFile is where the next release keeps the actor's snapshot +// state file: directly under ActorSnapshotPath. The file moves there over two +// releases (see docs/reference/version-skew-policy.md). atelet moves a file it +// finds here back to ActorSnapshotStateFile at startup, so a rollback from the +// next release keeps the actor's snapshots. +func NextSnapshotStateFile(actorUID string) string { + return filepath.Join( + ActorSnapshotPath(actorUID), + snapshotStateFileName, + ) +} + +const snapshotStateFileName = "snapshot-state.json" + // SandboxAssetsDir holds the actor's sandbox runtime assets, one entry per // asset name, each a hard link (or a directory of hard links, for a gVisor // release) from SandboxAssetCacheDir. The worker runs binaries from here diff --git a/cmd/atelet/internal/ateletpath/ateletpath_test.go b/cmd/atelet/internal/ateletpath/ateletpath_test.go index 74f9a54f23..9c6a0b4599 100644 --- a/cmd/atelet/internal/ateletpath/ateletpath_test.go +++ b/cmd/atelet/internal/ateletpath/ateletpath_test.go @@ -40,12 +40,14 @@ func TestActorDirs(t *testing.T) { "system_info_volume_roots_dir": actorDirs.GetSystemInfoVolumeRootsDir(), "external_volume_mounts_dir": actorDirs.GetExternalVolumeMountsDir(), "sandbox_assets_dir": actorDirs.GetSandboxAssetsDir(), + "snapshot_state_file": ActorSnapshotStateFile(actorUID), } underSnapshotRoot := map[string]string{ - "checkpoint_dir": actorDirs.GetCheckpointDir(), - "restore_dir": actorDirs.GetRestoreDir(), - "local_checkpoints_dir": LocalCheckpointsDir(actorUID), - "local_snapshot_dir": LocalSnapshotDir(actorUID, "snap-1"), + "checkpoint_dir": actorDirs.GetCheckpointDir(), + "restore_dir": actorDirs.GetRestoreDir(), + "local_checkpoints_dir": LocalCheckpointsDir(actorUID), + "local_snapshot_dir": LocalSnapshotDir(actorUID, "snap-1"), + "next_snapshot_state_file": NextSnapshotStateFile(actorUID), } seen := map[string]string{} diff --git a/cmd/atelet/internal/snapshottracker/state.go b/cmd/atelet/internal/snapshottracker/state.go index 02507a25e9..1e3fd7f9ac 100644 --- a/cmd/atelet/internal/snapshottracker/state.go +++ b/cmd/atelet/internal/snapshottracker/state.go @@ -20,6 +20,7 @@ import ( "fmt" "log/slog" "os" + "path/filepath" "syscall" "go.ate.dev/substrate/cmd/atelet/internal/ateletpath" @@ -120,3 +121,65 @@ func savedActorUIDs() ([]string, error) { } return actorUIDs, nil } + +// MigrateSnapshotStateFiles moves each actor's snapshot state file from +// ateletpath.NextSnapshotStateFile, where a future release keeps it, to +// ateletpath.ActorSnapshotStateFile. atelet calls it once at startup, before +// it creates the tracker: only one atelet runs on a node, so a file there was +// written before this atelet started, and from then on only +// ActorSnapshotStateFile is used. It syncs each moved file before removing the +// original, so a crash in between leaves two identical copies, which the next +// startup moves again. A file whose actor directory is gone is removed, as +// saveLocked keeps no state for such an actor. The actor's snapshot directory +// is then removed too if that left it empty, since nothing else removes it once +// the actor has detached or is gone. +func MigrateSnapshotStateFiles() error { + entries, err := os.ReadDir(nodepath.SnapshotsDir) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("while listing %s: %w", nodepath.SnapshotsDir, err) + } + for _, entry := range entries { + if !entry.IsDir() { + continue + } + if err := migrateSnapshotStateFile(entry.Name()); err != nil { + return err + } + } + return nil +} + +func migrateSnapshotStateFile(actorUID string) error { + from := ateletpath.NextSnapshotStateFile(actorUID) + data, err := os.ReadFile(from) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("while reading %s: %w", from, err) + } + if _, err := os.Stat(ateletpath.ActorPath(actorUID)); errors.Is(err, os.ErrNotExist) { + slog.Warn("Removing the snapshot state file of an actor whose directory is gone", slog.String("actorUID", actorUID), slog.String("path", from)) + } else if err != nil { + return fmt.Errorf("while checking the actor directory: %w", err) + } else { + to := ateletpath.ActorSnapshotStateFile(actorUID) + if err := os.MkdirAll(filepath.Dir(to), 0o700); err != nil { + return fmt.Errorf("while creating %s: %w", filepath.Dir(to), err) + } + if err := atomicfile.WriteSync(to, data, 0o600); err != nil { + return fmt.Errorf("while writing %s: %w", to, err) + } + slog.Info("Moved the actor's snapshot state file", slog.String("actorUID", actorUID), slog.String("from", from), slog.String("to", to)) + } + if err := os.Remove(from); err != nil { + return fmt.Errorf("while removing %s: %w", from, err) + } + if err := os.Remove(ateletpath.ActorSnapshotPath(actorUID)); err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, syscall.ENOTEMPTY) && !errors.Is(err, syscall.EEXIST) { + slog.Warn("Failed to remove the actor's empty snapshot directory", slog.String("actorUID", actorUID), slog.Any("err", err)) + } + return nil +} diff --git a/cmd/atelet/internal/snapshottracker/state_test.go b/cmd/atelet/internal/snapshottracker/state_test.go new file mode 100644 index 0000000000..c525813cb8 --- /dev/null +++ b/cmd/atelet/internal/snapshottracker/state_test.go @@ -0,0 +1,176 @@ +// Copyright 2026 The Agent Substrate Authors +// +// 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 snapshottracker + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + "go.ate.dev/substrate/cmd/atelet/internal/ateletpath" + "go.ate.dev/substrate/internal/proto/ateletpb" + "google.golang.org/protobuf/proto" +) + +// writeNextSnapshotState writes, where a future release keeps it, a snapshot +// state file recording local snapshot uuid for the tracked actor, and returns +// its contents. +func writeNextSnapshotState(t *testing.T, uuid string) []byte { + t.Helper() + ref := testRef(uuid, 1, time.Now()) + data, err := json.Marshal(actorSnapshotsRecord{LatestLocal: &ref}) + if err != nil { + t.Fatal(err) + } + path := ateletpath.NextSnapshotStateFile(trackedActorUID) + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + return data +} + +func assertNotExist(t *testing.T, path string) { + t.Helper() + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("stat(%s) = %v, want not exist", path, err) + } +} + +// A state file a future release wrote is moved to where this release keeps it, +// so the actor keeps its snapshots after a rollback. When both copies exist, as +// after a crash partway through an earlier move, the move is finished. The +// snapshot directory goes if the move left it empty, as for an actor that had +// detached, and stays if it holds anything else. +func TestMigrateSnapshotStateFiles(t *testing.T) { + for _, tc := range []struct { + name string + bothCopy bool + // otherFile is a file left in the snapshot directory, if not empty. + otherFile string + }{ + {name: "moves the file"}, + {name: "finishes an interrupted move", bothCopy: true}, + {name: "keeps a snapshot directory that holds other files", otherFile: "restore-state"}, + } { + t.Run(tc.name, func(t *testing.T) { + useTempNodeDirs(t) + if err := os.MkdirAll(ateletpath.ActorPath(trackedActorUID), 0o700); err != nil { + t.Fatal(err) + } + want := writeNextSnapshotState(t, "s1") + if tc.otherFile != "" { + if err := os.MkdirAll(filepath.Join(ateletpath.ActorSnapshotPath(trackedActorUID), tc.otherFile), 0o700); err != nil { + t.Fatal(err) + } + } + to := ateletpath.ActorSnapshotStateFile(trackedActorUID) + if tc.bothCopy { + if err := os.MkdirAll(filepath.Dir(to), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(to, want, 0o600); err != nil { + t.Fatal(err) + } + } + + if err := MigrateSnapshotStateFiles(); err != nil { + t.Fatalf("MigrateSnapshotStateFiles() = %v", err) + } + if got, err := os.ReadFile(to); err != nil || string(got) != string(want) { + t.Errorf("%s = (%q, %v), want (%q, nil)", to, got, err, want) + } + assertNotExist(t, ateletpath.NextSnapshotStateFile(trackedActorUID)) + if tc.otherFile == "" { + assertNotExist(t, ateletpath.ActorSnapshotPath(trackedActorUID)) + } else if _, err := os.Stat(filepath.Join(ateletpath.ActorSnapshotPath(trackedActorUID), tc.otherFile)); err != nil { + t.Errorf("snapshot directory entry %s: %v, want kept", tc.otherFile, err) + } + if got := uuidOf(newTracker(t).State(trackedActorUID).GetLatestLocal()); got != "s1" { + t.Errorf("latest_local = %q, want s1", got) + } + }) + } +} + +// A state file whose actor directory is gone, as after a terminate that a +// crash interrupted, is removed rather than moved, so the actor's state does +// not come back. +func TestMigrateSnapshotStateFilesRemovesFileOfGoneActor(t *testing.T) { + useTempNodeDirs(t) + writeNextSnapshotState(t, "s1") + + if err := MigrateSnapshotStateFiles(); err != nil { + t.Fatalf("MigrateSnapshotStateFiles() = %v", err) + } + assertNotExist(t, ateletpath.NextSnapshotStateFile(trackedActorUID)) + assertNotExist(t, ateletpath.ActorSnapshotPath(trackedActorUID)) + assertNotExist(t, ateletpath.ActorPath(trackedActorUID)) + if got := newTracker(t).State(trackedActorUID); !proto.Equal(got, &ateletpb.ActorSnapshotState{}) { + t.Errorf("State() = %v, want empty", got) + } +} + +// A move that fails keeps the original file, so the actor's state is not lost, +// and returns the error, so atelet does not start without it. +func TestMigrateSnapshotStateFilesKeepsFileWhenMoveFails(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root ignores directory permissions, so no write can be made to fail") + } + useTempNodeDirs(t) + if err := os.MkdirAll(ateletpath.ActorPath(trackedActorUID), 0o700); err != nil { + t.Fatal(err) + } + want := writeNextSnapshotState(t, "s1") + // A read-only destination directory makes the write fail. + toDir := filepath.Dir(ateletpath.ActorSnapshotStateFile(trackedActorUID)) + if err := os.MkdirAll(toDir, 0o700); err != nil { + t.Fatal(err) + } + if err := os.Chmod(toDir, 0o500); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.Chmod(toDir, 0o700) }) + + if err := MigrateSnapshotStateFiles(); err == nil { + t.Fatal("MigrateSnapshotStateFiles() = nil, want an error") + } + from := ateletpath.NextSnapshotStateFile(trackedActorUID) + if got, err := os.ReadFile(from); err != nil || string(got) != string(want) { + t.Errorf("%s = (%q, %v), want (%q, nil)", from, got, err, want) + } +} + +// A node with nothing to move needs no change. +func TestMigrateSnapshotStateFilesWithNothingToMove(t *testing.T) { + useTempNodeDirs(t) + if err := MigrateSnapshotStateFiles(); err != nil { + t.Fatalf("MigrateSnapshotStateFiles() = %v, want nil on a node with no directories", err) + } + if err := os.MkdirAll(ateletpath.LocalCheckpointsDir(trackedActorUID), 0o700); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(ateletpath.ActorPath(trackedActorUID), 0o700); err != nil { + t.Fatal(err) + } + if err := MigrateSnapshotStateFiles(); err != nil { + t.Fatalf("MigrateSnapshotStateFiles() = %v, want nil with no state file to move", err) + } + assertNotExist(t, ateletpath.ActorSnapshotStateFile(trackedActorUID)) +} diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 42d1e4d75b..74f34b8ac6 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -401,6 +401,9 @@ func main() { defer ateapiConn.Close() workerService := ateapi.NewWorkerServiceClient(ateapiConn) + if err := snapshottracker.MigrateSnapshotStateFiles(); err != nil { + serverboot.Fatal(ctx, "Failed to migrate the actors' snapshot state files", err) + } // The snapshot tracker's workers get a context of their own, cancelled only // once the gRPC server has drained, so background snapshot work keeps // running while in-flight RPCs finish.