atelet: clear node-local caches after a node reboot

cmd/atelet/bootid.go

@@ -15,7 +15,16 @@
 package main
 
 import (
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strings"
+
 	"go.ate.dev/substrate/cmd/atelet/internal/ateletpath"
+	"go.ate.dev/substrate/cmd/atelet/internal/atomicfile"
+	"go.ate.dev/substrate/internal/imagecache"
 )
 
 // localSnapshotBootCheck returns the snapshot tracker's check of whether an
@@ -32,3 +41,75 @@ func localSnapshotBootCheck(bootID string) func(actorUID, snapshotUUID string) b
 		return err == nil && rec.BootID == bootID
 	}
 }
+
+// clearCachesAfterReboot empties each of cacheDirs, and removes every actor's
+// OCI bundle dir and every worker pod's default-runtime dir, when markerPath
+// records a boot other than bootID. Then it records bootID there. It reports
+// whether it cleared anything.
+//
+// Cache writes aren't synced, so an unclean reboot can leave cache entries
+// and bundle specs empty or truncated. The bundle and default-runtime dirs
+// hold specs naming cached layers and hard links to cached assets, which
+// atelet reuses instead of linking again. No sandbox survives a reboot, so
+// nothing still uses any of them. Run recreates an actor's bundle dir, and the
+// worker sandbox syncer recreates a worker's default-runtime dir.
+//
+// A missing marker only records bootID. atelet may be starting with this
+// check for the first time while actors run on what it would clear.
+func clearCachesAfterReboot(bootID, markerPath, actorsDir, workersDir string, cacheDirs ...string) (bool, error) {
+	b, err := os.ReadFile(markerPath)
+	switch {
+	case errors.Is(err, fs.ErrNotExist):
+		return false, writeCacheBootID(markerPath, bootID)
+	case err != nil:
+		return false, fmt.Errorf("while reading the cache boot ID: %w", err)
+	case strings.TrimSpace(string(b)) == bootID:
+		return false, nil
+	}
+	for _, dir := range cacheDirs {
+		if err := removeChildren(dir, ""); err != nil {
+			return false, err
+		}
+	}
+	if err := removeChildren(actorsDir, "bundles"); err != nil {
+		return false, err
+	}
+	if err := removeChildren(workersDir, defaultRuntimeDirsName); err != nil {
+		return false, err
+	}
+	return true, writeCacheBootID(markerPath, bootID)
+}
+
+// removeChildren removes the named entry in each child directory of dir, or
+// each child itself when name is empty. A missing dir has nothing to remove.
+func removeChildren(dir, name string) error {
+	if dir == "" {
+		return nil
+	}
+	children, err := os.ReadDir(dir)
+	if errors.Is(err, fs.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("while listing %s: %w", dir, err)
+	}
+	for _, child := range children {
+		if name != "" && !child.IsDir() {
+			continue
+		}
+		path := filepath.Join(dir, child.Name(), name)
+		// RemoveAllWritable, not os.RemoveAll: unpacked image layers keep
+		// the image's directory modes, which can be read-only.
+		if err := imagecache.RemoveAllWritable(path); err != nil {
+			return fmt.Errorf("while removing %s: %w", path, err)
+		}
+	}
+	return nil
+}
+
+func writeCacheBootID(path, bootID string) error {
+	if err := atomicfile.WriteSync(path, []byte(bootID+"\n"), 0o600); err != nil {
+		return fmt.Errorf("while recording the cache boot ID: %w", err)
+	}
+	return nil
+}

cmd/atelet/bootid_test.go

@@ -15,6 +15,8 @@
 package main
 
 import (
+	"errors"
+	"io/fs"
 	"os"
 	"path/filepath"
 	"testing"
@@ -59,3 +61,87 @@ func TestLocalSnapshotBootCheck(t *testing.T) {
 		})
 	}
 }
+
+// clearCachesAfterReboot clears the caches, the actors' bundle dirs and the
+// workers' default-runtime dirs only once the node has rebooted since the
+// recorded boot, and leaves everything else in place.
+func TestClearCachesAfterReboot(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		// marker is the recorded boot ID, or empty for no marker file.
+		marker      string
+		wantCleared bool
+	}{
+		{name: "rebooted since the recorded boot", marker: "boot-1\n", wantCleared: true},
+		{name: "same boot", marker: "boot-2\n"},
+		{name: "no recorded boot", marker: ""},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			root := t.TempDir()
+			marker := filepath.Join(root, "cache-boot-id")
+			imageCache := filepath.Join(root, "image-cache")
+			assetCache := filepath.Join(root, "sandbox-asset-cache")
+			actors := filepath.Join(root, "actors")
+			workers := filepath.Join(root, "workers")
+
+			cleared := []string{
+				filepath.Join(imageCache, "layers", "sha256", "abc", "fs", "etc", "ca.crt"),
+				filepath.Join(imageCache, "version"),
+				filepath.Join(assetCache, "entries", "k1", "data"),
+				filepath.Join(actors, "a1", "bundles", "c1", "overlay.json"),
+				filepath.Join(workers, "w1", "default-runtime", "sha256-x", "runsc"),
+			}
+			kept := []string{
+				filepath.Join(actors, "a1", "snapshot-state.json"),
+				filepath.Join(actors, "a1", "sandbox-assets.json"),
+				filepath.Join(workers, "w1", "sandbox-runtimes.json"),
+				filepath.Join(root, "snapshots", "a1", "local-checkpoint"),
+			}
+			for _, p := range append(append([]string{}, cleared...), kept...) {
+				if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil {
+					t.Fatal(err)
+				}
+				if err := os.WriteFile(p, nil, 0o600); err != nil {
+					t.Fatal(err)
+				}
+			}
+			// A plain file directly in the actors dir has no bundle dir.
+			if err := os.WriteFile(filepath.Join(actors, "stray"), nil, 0o600); err != nil {
+				t.Fatal(err)
+			}
+			// Unpacked layers keep the image's directory modes.
+			readOnly := filepath.Join(imageCache, "layers", "sha256", "abc", "fs", "etc")
+			if err := os.Chmod(readOnly, 0o500); err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { _ = os.Chmod(readOnly, 0o700) })
+			if tc.marker != "" {
+				if err := os.WriteFile(marker, []byte(tc.marker), 0o600); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			got, err := clearCachesAfterReboot("boot-2", marker, actors, workers, imageCache, assetCache, "")
+			if err != nil {
+				t.Fatalf("clearCachesAfterReboot: %v", err)
+			}
+			if got != tc.wantCleared {
+				t.Errorf("clearCachesAfterReboot cleared = %v, want %v", got, tc.wantCleared)
+			}
+			for _, p := range cleared {
+				_, err := os.Stat(p)
+				if gone := errors.Is(err, fs.ErrNotExist); gone != tc.wantCleared {
+					t.Errorf("%s removed = %v, want %v (stat err %v)", p, gone, tc.wantCleared, err)
+				}
+			}
+			for _, p := range append(kept, imageCache, assetCache, filepath.Join(actors, "stray")) {
+				if _, err := os.Stat(p); err != nil {
+					t.Errorf("%s should stay: %v", p, err)
+				}
+			}
+			if b, err := os.ReadFile(marker); err != nil || string(b) != "boot-2\n" {
+				t.Errorf("marker = (%q, %v), want (%q, nil)", b, err, "boot-2\n")
+			}
+		})
+	}
+}

cmd/atelet/internal/ateletpath/ateletpath.go

@@ -57,6 +57,11 @@ var (
 	// again.
 	SandboxRuntimeIndexFile = filepath.Join(nodepath.BasePath, "sandbox-runtime-index.json")
 
+	// CacheBootIDFile records the boot ID of the boot that last wrote atelet's
+	// caches. atelet clears them at startup when the node has rebooted since
+	// (see clearCachesAfterReboot in package main).
+	CacheBootIDFile = filepath.Join(nodepath.BasePath, "cache-boot-id")
+
 	// SnapshotCacheDir is the node-local cache of shared snapshot files, the
 	// snapshots many actors are created from (see
 	// cmd/atelet/internal/filecache). It lives under BasePath because cache

cmd/atelet/main.go

@@ -254,9 +254,30 @@ func main() {
 		imageCredsKeychain = kc
 	}
 
+	// Validated before the caches are cleared, so a directory the flags
+	// reject is never emptied.
 	if err := validateImageCacheGCFlags(); err != nil {
 		serverboot.Fatal(ctx, "Invalid image cache GC flags", err)
 	}
+	if err := validateSandboxAssetCacheFlags(); err != nil {
+		serverboot.Fatal(ctx, "Invalid sandbox asset cache flags", err)
+	}
+	if err := validateSnapshotCacheFlags(); err != nil {
+		serverboot.Fatal(ctx, "Invalid snapshot cache flags", err)
+	}
+	bootID, err := bootid.Read(bootid.Path)
+	if err != nil {
+		serverboot.Fatal(ctx, "Failed to read the node's boot ID", err)
+	}
+	cleared, err := clearCachesAfterReboot(bootID, ateletpath.CacheBootIDFile, nodepath.ActorsDir, nodepath.WorkersDir(),
+		*imageCacheDir, *sandboxAssetCacheDir, *snapshotCacheDir)
+	if err != nil {
+		serverboot.Fatal(ctx, "Failed to clear the caches after a reboot", err)
+	}
+	if cleared {
+		slog.InfoContext(ctx, "Cleared the node-local caches after a node reboot")
+	}
+
 	imageCache, err := imagecache.New(*imageCacheDir,
 		imagecache.WithKeychain(imageCredsKeychain),
 		imagecache.WithLocalhostRegistryReplacement(*localhostRegistryReplacement),
@@ -271,9 +292,6 @@ func main() {
 		go newImageCacheGC(imageCache, *imageCacheDir).Run(ctx)
 	}
 
-	if err := validateSandboxAssetCacheFlags(); err != nil {
-		serverboot.Fatal(ctx, "Invalid sandbox asset cache flags", err)
-	}
 	sandboxAssetCache, err := openSandboxAssetCache(ctx, *sandboxAssetCacheDir, *sandboxAssetCacheMinAge, *sandboxAssetFetchTimeout, nodepath.ActorsDir, nodepath.WorkersDir())
 	if err != nil {
 		serverboot.Fatal(ctx, "Failed to open sandbox asset cache", err)
@@ -286,9 +304,6 @@ func main() {
 	if err := filecachegc.Start(ctx, ateattr.FileCacheSandboxAsset, sandboxAssetCache, sandboxAssetCacheGCFlags, fileCacheInstruments); err != nil {
 		serverboot.Fatal(ctx, "Failed to start the sandbox asset cache GC", err)
 	}
-	if err := validateSnapshotCacheFlags(); err != nil {
-		serverboot.Fatal(ctx, "Invalid snapshot cache flags", err)
-	}
 	snapshotCache, err := openSnapshotCache(ctx, *snapshotCacheDir, *snapshotCacheMinAge, *snapshotCacheFetchTimeout)
 	if err != nil {
 		serverboot.Fatal(ctx, "Failed to open snapshot cache", err)
@@ -385,10 +400,6 @@ func main() {
 	defer ateapiConn.Close()
 	workerService := ateapi.NewWorkerServiceClient(ateapiConn)
 
-	bootID, err := bootid.Read(bootid.Path)
-	if err != nil {
-		serverboot.Fatal(ctx, "Failed to read the node's boot ID", 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.

docs/sandbox-asset-cache.md

@@ -28,6 +28,7 @@ cannot cross mounts.
 | `workers/<pod uid>/default-runtime/<hash>/` | The worker's default runtimes, one directory per default its file advertises, by bundle hash, one per sandbox class, linked in from the cache by the prewarmer. These links are what keeps a default runtime cached while no actor runs it; the syncer removes a directory when the file stops naming its runtime, and they all go with the worker's directory when the pod does. |
 | `sandbox-asset-scratch/` | Short-lived destinations for fetches that have no actor and no worker directory to link into: the assets the worker sandbox syncer stages when checking a version (and probes `runsc` for CPU features). Wiped at startup. |
 | `sandbox-runtime-index.json` | The runtime index serialized: every runtime a worker on this node has advertised, by hash, with the asset record the hash was computed from. |
+| `cache-boot-id` | The node's boot ID when `atelet` last started. A different ID at startup means the node rebooted, so `atelet` empties the cache first. |
 
 A cache entry is keyed by the asset's sha256, so the same binaries under
 another name or from another bucket are one entry. The store never
@@ -129,6 +130,9 @@ which directories exist. Nothing counts references and nothing prunes.
   restart the directories of pods still on the node keep protecting their
   defaults with no wait on the syncer, and the startup sweep removes the
   directories of pods that are gone.
+- Cache writes aren't synced, so after a node reboot atelet empties the cache
+  and removes every worker's `default-runtime/` directory at startup. The
+  syncer and the prewarmer then rebuild them as on a new node.
 
 The old default's directory is removed as soon as the file names the new
 default: by then the worker has registered the new default (and unless an

internal/imagecache/README.md

@@ -24,9 +24,10 @@ What it buys, concretely:
   single-digit milliseconds on warm nodes.
 - Layers shared between images are **downloaded and unpacked once per
   node**, not once per image; actors sharing layers also share page cache.
-- The cache is **on disk and survives atelet restarts and node reboots**
-  (the old in-memory cache was lost on every restart, and its unbounded heap
-  retention could OOM atelet — #437).
+- The cache is **on disk and survives atelet restarts** (the old in-memory
+  cache was lost on every restart, and its unbounded heap retention could OOM
+  atelet — #437). Its writes aren't synced, so atelet empties it at startup
+  after a node reboot.
 - **Tag refs are cacheable**: a tag costs one `HEAD` request to resolve to a
   manifest digest (the only safe cache key for mutable tags); digest refs
   hit the cache with zero network I/O.