atelet: serve shared snapshot restores from the node-local file cache

cmd/ateapi/internal/controlapi/functionaltest/actor_test.go

@@ -3365,6 +3365,11 @@ func TestResumeActor_DataSnapshotIgnoresGolden(t *testing.T) {
 	if got := restoreReq.GetSnapshot().GetObject().GetSnapshotUri(); got != actorSnapshotURI {
 		t.Errorf("restore config snapshot uri = %q, want the actor's data snapshot %q", got, actorSnapshotURI)
 	}
+	// The actor's own snapshot is replaced by its next suspend: atelet must
+	// not cache it.
+	if got := restoreReq.GetSnapshot().GetObject().GetSharing(); got != ateletpb.SnapshotSharing_SNAPSHOT_SHARING_PRIVATE {
+		t.Errorf("restore of the actor's own snapshot declared sharing %v, want PRIVATE", got)
+	}
 }
 
 // TestSuspendActor_ReplacedSnapshotReleaseFailure verifies that object storage
@@ -6466,6 +6471,11 @@ func TestResumeActor_SandboxRuntimePlacementAndFallback(t *testing.T) {
 	if got, want := goldenRestore.GetSnapshot().GetObject().GetSnapshotUri(), goldenSnapshotURI(t); got != want {
 		t.Fatalf("golden-actor RestoreRequest.SnapshotUri = %q, want golden %q", got, want)
 	}
+	// A borrowed golden snapshot is shared by every actor created from it,
+	// and the request must say so: atelet caches only declared-SHARED sources.
+	if got := goldenRestore.GetSnapshot().GetObject().GetSharing(); got != ateletpb.SnapshotSharing_SNAPSHOT_SHARING_SHARED {
+		t.Fatalf("golden-actor RestoreRequest.Sharing = %v, want SHARED", got)
+	}
 
 	// Free worker-b and have it advertise rtA in restorable_runtimes while
 	// keeping rtB as its default. Resuming an actor whose snapshot requires rtA

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -48,6 +48,11 @@ type resumeSnapshotSource struct {
 	// snapshot, which takes precedence at restore).
 	SnapshotURI resources.SnapshotURI
 	Fidelity    ateapipb.SnapshotFidelity
+	// Owner is who the durable snapshot belongs to: the actor, whose next
+	// suspend replaces it, or a tag, whose snapshot is immutable and read by
+	// every actor created from it. It decides what the restore declares as
+	// the snapshot's sharing.
+	Owner ateapipb.SnapshotOwner
 	// TemplateReplaced is true when the external snapshot's recorded template
 	// UID differs from the actor's current template.
 	TemplateReplaced bool
@@ -198,6 +203,7 @@ func resolveResumeSource(actorRef resources.ActorRef, actor *ateapipb.Actor, act
 		src.SnapshotUUID = snap.GetUuid()
 		src.Generation = snap.GetGeneration()
 		src.Fidelity = snap.GetFidelity()
+		src.Owner = snap.GetOwner()
 		src.TemplateReplaced = templateReplaced(snap, actorTemplate)
 	}
 	// A local restore also names the local snapshot's upload, so a URI that
@@ -217,6 +223,18 @@ func templateReplaced(snap *ateapipb.Snapshot, tmpl *ateapipb.ActorTemplate) boo
 	return capturedUnder != "" && capturedUnder != tmpl.GetMetadata().GetUid()
 }
 
+// snapshotSharing is what a restore declares about a snapshot with the
+// given owner. A tag's snapshot is immutable once published and read by
+// every actor created from it, so atelet may serve its files from the
+// node-local cache; the actor's own is replaced by its next suspend, so
+// never, and so is one whose owner is not recorded.
+func snapshotSharing(owner ateapipb.SnapshotOwner) ateletpb.SnapshotSharing {
+	if owner == ateapipb.SnapshotOwner_SNAPSHOT_OWNER_TAG {
+		return ateletpb.SnapshotSharing_SNAPSHOT_SHARING_SHARED
+	}
+	return ateletpb.SnapshotSharing_SNAPSHOT_SHARING_PRIVATE
+}
+
 // ensureVolumesCreated provisions any actor volumes that have no storage
 // volume ID yet, persisting the resulting volume state (even when creation
 // partially failed, so progress is not lost) and returning the stored copy.
@@ -827,7 +845,10 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou
 				// report of the upload lands. atelet reads it only when it
 				// holds no usable local copy, and the manifest, uploaded
 				// last, tells a complete upload from a partial one.
-				Object: &ateletpb.ObjectStorage{SnapshotUri: localSnap.GetDurableSnapshot().GetObject().GetSnapshotUri()},
+				Object: &ateletpb.ObjectStorage{
+					SnapshotUri: localSnap.GetDurableSnapshot().GetObject().GetSnapshotUri(),
+					Sharing:     snapshotSharing(localSnap.GetOwner()),
+				},
 			},
 			SandboxRuntime: sandboxRuntime,
 			ActorUid:       actor.GetMetadata().Uid,
@@ -868,6 +889,7 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou
 				Fidelity:     scope,
 				Object: &ateletpb.ObjectStorage{
 					SnapshotUri: src.SnapshotURI.String(),
+					Sharing:     snapshotSharing(src.Owner),
 				},
 			},
 			SandboxRuntime: sandboxRuntime,

cmd/ateapi/internal/controlapi/workflow_resume_detach_test.go

@@ -364,7 +364,6 @@ func TestResumeActor_DetachesFromFullNode(t *testing.T) {
 				a.Status.LastAssignedGeneration = 1
 				a.Status.Snapshots = []*ateapipb.Snapshot{proto.CloneOf(snap)}
 			})
-
 			w, atelet := newDetachTestWorkflow(t, ctx, st)
 			addAteletOnNode(t, w, "node-2")
 			atelet.detachState = detachedState(snap)
@@ -394,7 +393,12 @@ func TestResumeActor_DetachesFromFullNode(t *testing.T) {
 				SnapshotUuid: snap.GetUuid(),
 				Generation:   snap.GetGeneration(),
 				Fidelity:     ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY,
-				Object:       &ateletpb.ObjectStorage{SnapshotUri: snap.GetDurableSnapshot().GetObject().GetSnapshotUri()},
+				// The actor's own snapshot, so private, whether atelet
+				// downloads it or restores the local copy.
+				Object: &ateletpb.ObjectStorage{
+					SnapshotUri: snap.GetDurableSnapshot().GetObject().GetSnapshotUri(),
+					Sharing:     ateletpb.SnapshotSharing_SNAPSHOT_SHARING_PRIVATE,
+				},
 			}
 			if !proto.Equal(restore.GetSnapshot(), want) {
 				t.Errorf("Restore snapshot = %v, want %v", restore.GetSnapshot(), want)

cmd/atelet/imagegc.go

@@ -29,10 +29,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"os"
-	"path/filepath"
 	"runtime/debug"
-	"strings"
 	"time"
 
 	"github.com/spf13/pflag"
@@ -76,26 +73,17 @@ func validateImageCacheGCFlags() error {
 		// future), making just-pulled layers evictable.
 		return fmt.Errorf("--image-cache-min-age %v must be >= 0", *imageCacheMinAge)
 	}
-	if imageCacheDirOutsideBasePath(*imageCacheDir) {
+	// Warn-worthy, not an error: a separate cache volume is legitimate
+	// (recommended for IOPS), but the watermarks then measure a different
+	// volume than actor state.
+	if !nodepath.UnderBasePath(*imageCacheDir) {
 		slog.Warn("Image cache dir is outside the worker base path; its volume watermarks are measured separately from actor state",
 			slog.String("image_cache_dir", *imageCacheDir),
 			slog.String("actors_dir", nodepath.ActorsDir))
 	}
 	return nil
 }
 
-// imageCacheDirOutsideBasePath reports whether the cache dir is outside
-// the worker base path — the watermarks then measure a different volume
-// than actor state. Warn-worthy, not an error: a separate cache volume is
-// legitimate (recommended for IOPS).
-func imageCacheDirOutsideBasePath(dir string) bool {
-	abs, err := filepath.Abs(dir)
-	if err != nil {
-		abs = filepath.Clean(dir)
-	}
-	return !strings.HasPrefix(abs, nodepath.BasePath+string(os.PathSeparator))
-}
-
 // imageCacheGCTarget computes the bytes a pass should free: the larger
 // of the watermark shortfall (kubelet's formula — usage at highPct frees
 // down to lowPct) and the pool's overage past maxBytes, but never more

cmd/atelet/imagegc_test.go

@@ -26,7 +26,6 @@ import (
 	"time"
 
 	"go.ate.dev/substrate/internal/imagecache"
-	"go.ate.dev/substrate/internal/nodepath"
 )
 
 func TestImageCacheGCTarget(t *testing.T) {
@@ -154,30 +153,6 @@ func TestValidateImageCacheGCFlags(t *testing.T) {
 	}
 }
 
-func TestImageCacheDirOutsideBasePath(t *testing.T) {
-	cases := []struct {
-		name string
-		dir  string
-		want bool
-	}{
-		{"inside", filepath.Join(nodepath.BasePath, "image-cache"), false},
-		{"inside with doubled separator", nodepath.BasePath + "//image-cache", false},
-		{"inside via dot-dot", nodepath.BasePath + "/x/../image-cache", false},
-		{"base path itself is not inside", nodepath.BasePath, true},
-		{"sibling with the base path as name prefix", nodepath.BasePath + "-other/image-cache", true},
-		{"outside", "/var/lib/elsewhere/image-cache", true},
-		{"dot-dot escaping the base path", nodepath.BasePath + "/../elsewhere/image-cache", true},
-		{"relative resolves against the cwd, not the base path", "image-cache", true},
-	}
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			if got := imageCacheDirOutsideBasePath(tc.dir); got != tc.want {
-				t.Errorf("imageCacheDirOutsideBasePath(%q) = %v, want %v", tc.dir, got, tc.want)
-			}
-		})
-	}
-}
-
 type fakeGCStore struct {
 	size         int64
 	sizeErr      error

cmd/atelet/internal/ateletpath/ateletpath.go

@@ -55,6 +55,14 @@ var (
 	// runtime registered before a restart resolves the moment atelet serves
 	// again.
 	SandboxRuntimeIndexFile = filepath.Join(nodepath.BasePath, "sandbox-runtime-index.json")
+
+	// 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
+	// hits are served as hard links into the per-actor restore dirs under
+	// ActorsDir, and link(2) requires both ends on one mounted filesystem,
+	// the same mount, not merely the same disk.
+	SnapshotCacheDir = filepath.Join(nodepath.BasePath, "snapshot-cache")
 )
 
 func ActorPath(actorUID string) string {

cmd/atelet/lifecycle_test.go

@@ -30,6 +30,7 @@ import (
 	"time"
 
 	"go.ate.dev/substrate/cmd/atelet/internal/ateletpath"
+	"go.ate.dev/substrate/cmd/atelet/internal/filecache"
 	"go.ate.dev/substrate/cmd/atelet/internal/snapshottracker"
 	"go.ate.dev/substrate/cmd/atelet/internal/trustbundle"
 	"go.ate.dev/substrate/internal/apierror"
@@ -76,6 +77,9 @@ type fakeWorker struct {
 	// restored holds the file contents staged into the restore dir by the
 	// most recent RestoreWorkload.
 	restored map[string]string
+	// restoreErr, if set, fails every RestoreWorkload after it recorded the
+	// request, as a worker that could not start the sandbox would.
+	restoreErr error
 	// actorDirs records the ActorDirs each RPC arrived with, by RPC name.
 	actorDirs map[string]*ateworkerpb.ActorDirs
 	// preserveRestoreDir records the PreserveRestoreDir flag from the most
@@ -136,6 +140,9 @@ func (f *fakeWorker) RestoreWorkload(_ context.Context, req *ateworkerpb.Restore
 	f.recordActorDirs("RestoreWorkload", req.GetActorDirs())
 	f.recordFidelity("RestoreWorkload", req.GetFidelity())
 	f.preserveRestoreDir = req.GetPreserveRestoreDir()
+	if f.restoreErr != nil {
+		return nil, f.restoreErr
+	}
 	dir := req.GetActorDirs().GetRestoreDir()
 	f.restored = map[string]string{}
 	for name := range f.snapshotFiles {
@@ -897,6 +904,86 @@ func TestRestorePrefersLocalAndFallsBackToObjectStorage(t *testing.T) {
 	}
 }
 
+// TestRestoreFailureRemovesStagedSnapshotFiles pins what a shared restore
+// that fails after its snapshot files were staged leaves behind: nothing in
+// the restore-state directory, so the links into the snapshot cache do not
+// hold its entries until a later lifecycle call resets the directory.
+func TestRestoreFailureRemovesStagedSnapshotFiles(t *testing.T) {
+	useTempNodeDirs(t)
+	const (
+		actorUID     = "0f6c2b0e-1b7e-4b3f-9a6a-1d2c3e4f5a66"
+		snapshotName = "pause-snap-1"
+	)
+	worker := &fakeWorker{snapshotFiles: map[string]string{"checkpoint.img": ""}, restoreErr: status.Error(codes.Internal, "sandbox would not start")}
+	serveFakeWorker(t, worker)
+	host := imageVolumeTestRegistry(t)
+	pinnedImage := pushTestImage(t, host+"/actor:v1", singleFileLayer(t, "bin/app", "app"))
+	release := newGVisorTestRelease(t)
+	sandboxRec := testSandboxRecord(release, pinnedImage)
+	manifestRec := sandboxAssetsRecord{
+		SandboxClass:  "gvisor",
+		PauseImage:    pinnedImage,
+		Assets:        map[string]assetEntry{gvisorAssetName: release.entry},
+		SnapshotFiles: []string{"checkpoint.img"},
+	}
+	s := newPluginHerder(t, &recordingObjectStorage{})
+	s.workerDialer = newWorkerDialer(1)
+	s.imageCache = newImageVolumeStore(t)
+	s.anonGCSClient = fakeObjectStorage{data: release.tarball}
+	s.systemInfoVolumes = newSystemInfoVolumeRefresher(nil, nil)
+	// No min age, so the only thing standing between a staged entry and a
+	// pass is the restore's link.
+	cache, err := filecache.New(filepath.Join(t.TempDir(), "cache"), filecache.WithMinAge(0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.snapshotCache = cache
+	uri, err := resources.ParseSnapshotURI(pausedSnapshotURI)
+	if err != nil {
+		t.Fatal(err)
+	}
+	srcDir := t.TempDir()
+	writeLocalSnapshot(t, srcDir, manifestRec, map[string]string{"checkpoint.img": "shared-memory"})
+	if err := s.uploadSnapshot(t.Context(), uri, srcDir, &manifestRec); err != nil {
+		t.Fatalf("uploadSnapshot: %v", err)
+	}
+	advertiseRuntime(t, s, sandboxRec)
+
+	_, err = s.Restore(t.Context(), &ateletpb.RestoreRequest{
+		Atespace:              "ate-demo",
+		ActorName:             "counter",
+		ActorUid:              actorUID,
+		ActorTemplateAtespace: "default",
+		ActorTemplateName:     "counter",
+		WorkerPodUid:          "5f1c0d2e-3a4b-4c5d-8e9f-0a1b2c3d4e5f",
+		SandboxRuntime:        runtimeFor(t, sandboxRec),
+		Spec:                  &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "app", Image: pinnedImage, Command: []string{"/bin/app"}}}},
+		Snapshot: &ateletpb.Snapshot{
+			SnapshotUuid: snapshotName,
+			Generation:   1,
+			Fidelity:     ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY,
+			Object:       &ateletpb.ObjectStorage{SnapshotUri: pausedSnapshotURI, Sharing: ateletpb.SnapshotSharing_SNAPSHOT_SHARING_SHARED},
+		},
+	})
+	if status.Code(err) != codes.Internal {
+		t.Fatalf("Restore = %v, want the worker's Internal", err)
+	}
+	if worker.actorDirs["RestoreWorkload"] == nil {
+		t.Fatal("the worker was never asked: the failure must come after the snapshot files were staged")
+	}
+	if entries, err := os.ReadDir(ateletpath.RestoreStateDir(actorUID)); err == nil && len(entries) != 0 {
+		t.Errorf("restore-state holds %v after a failed restore, want nothing", entries)
+	}
+	stats, err := cache.EvictUnused(t.Context(), 1<<40, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Two entries, the manifest and checkpoint.img, both unlinked.
+	if stats.Retired != 2 || stats.SkippedLinked != 0 {
+		t.Errorf("after the failed restore a pass would retire %d entries and skip %d linked, want 2 and 0: the staged link must be gone", stats.Retired, stats.SkippedLinked)
+	}
+}
+
 // TestRestoreWithoutUsableLocalCopy checks that a Restore with no usable local
 // copy, naming a snapshot.object that was never uploaded, fails as NotFound
 // and wraps the local copy's problem.