Add ActorDirs message to carry per-actor directories

cmd/atelet/lifecycle_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
 	"github.com/agent-substrate/substrate/internal/proto/ateompb"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
 )
 
 // useTempNodeDirs roots atelet's on-node state in temp directories so a test
@@ -45,24 +46,36 @@ func useTempNodeDirs(t *testing.T) {
 }
 
 // fakeAteom is a fake ateom in a worker pod. It writes the files a
-// real checkpoint would leave in the checkpoint-state dir, and reads back
-// what a restore was handed.
+// real checkpoint would leave in the checkpoint dir, and reads back what a
+// restore was handed. Like a real ateom it takes every actor directory from
+// the request, never from ateompath.
 type fakeAteom struct {
 	ateompb.UnimplementedAteomServer
 	// snapshotFiles are written at checkpoint and reported back to atelet as
 	// the exact set the snapshot consists of.
 	snapshotFiles map[string]string
-	// restored holds the file contents staged into the restore-state dir by
-	// the most recent RestoreWorkload.
+	// restored holds the file contents staged into the restore dir by the
+	// most recent RestoreWorkload.
 	restored map[string]string
+	// dirs records the ActorDirs each RPC arrived with, by RPC name.
+	dirs map[string]*ateompb.ActorDirs
 }
 
-func (f *fakeAteom) RunWorkload(context.Context, *ateompb.RunWorkloadRequest) (*ateompb.RunWorkloadResponse, error) {
+func (f *fakeAteom) recordDirs(rpc string, dirs *ateompb.ActorDirs) {
+	if f.dirs == nil {
+		f.dirs = map[string]*ateompb.ActorDirs{}
+	}
+	f.dirs[rpc] = dirs
+}
+
+func (f *fakeAteom) RunWorkload(_ context.Context, req *ateompb.RunWorkloadRequest) (*ateompb.RunWorkloadResponse, error) {
+	f.recordDirs("RunWorkload", req.GetActorDirs())
 	return &ateompb.RunWorkloadResponse{}, nil
 }
 
 func (f *fakeAteom) CheckpointWorkload(_ context.Context, req *ateompb.CheckpointWorkloadRequest) (*ateompb.CheckpointWorkloadResponse, error) {
-	dir := ateompath.CheckpointStateDir(req.GetActorUid())
+	f.recordDirs("CheckpointWorkload", req.GetActorDirs())
+	dir := req.GetActorDirs().GetCheckpointDir()
 	names := make([]string, 0, len(f.snapshotFiles))
 	for name, body := range f.snapshotFiles {
 		if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil {
@@ -74,7 +87,8 @@ func (f *fakeAteom) CheckpointWorkload(_ context.Context, req *ateompb.Checkpoin
 }
 
 func (f *fakeAteom) RestoreWorkload(_ context.Context, req *ateompb.RestoreWorkloadRequest) (*ateompb.RestoreWorkloadResponse, error) {
-	dir := ateompath.RestoreStateDir(req.GetActorUid())
+	f.recordDirs("RestoreWorkload", req.GetActorDirs())
+	dir := req.GetActorDirs().GetRestoreDir()
 	f.restored = map[string]string{}
 	for name := range f.snapshotFiles {
 		body, err := os.ReadFile(filepath.Join(dir, name))
@@ -86,7 +100,8 @@ func (f *fakeAteom) RestoreWorkload(_ context.Context, req *ateompb.RestoreWorkl
 	return &ateompb.RestoreWorkloadResponse{}, nil
 }
 
-func (f *fakeAteom) TerminateWorkload(context.Context, *ateompb.TerminateWorkloadRequest) (*ateompb.TerminateWorkloadResponse, error) {
+func (f *fakeAteom) TerminateWorkload(_ context.Context, req *ateompb.TerminateWorkloadRequest) (*ateompb.TerminateWorkloadResponse, error) {
+	f.recordDirs("TerminateWorkload", req.GetActorDirs())
 	return &ateompb.TerminateWorkloadResponse{}, nil
 }
 
@@ -231,6 +246,15 @@ func TestLocalSnapshotGC(t *testing.T) {
 		t.Fatalf("Terminate: %v", err)
 	}
 
+	// Every RPC hands ateom the same directory set; the fake already relied
+	// on checkpoint_dir and restore_dir above to place and find the snapshot.
+	want := actorDirsFor(actorUID)
+	for _, rpc := range []string{"RunWorkload", "CheckpointWorkload", "RestoreWorkload", "TerminateWorkload"} {
+		if got := ateom.dirs[rpc]; !proto.Equal(got, want) {
+			t.Errorf("%s carried actor dirs %v, want %v", rpc, got, want)
+		}
+	}
+
 	localDir := ateompath.LocalCheckpointsDir(actorUID)
 	if _, err := os.Stat(localDir); !os.IsNotExist(err) {
 		leaked, _ := filepath.Glob(filepath.Join(localDir, "*", "*"))

cmd/atelet/main.go

@@ -539,6 +539,7 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp *
 		RuntimeAssetPaths:     assetPaths,
 		Spec:                  spec,
 		ActorUid:              actorUID,
+		ActorDirs:             actorDirsFor(actorUID),
 		EgressGateway:         toAteomEgressGateway(req.GetEgressGateway()),
 		CpuMilli:              req.GetCpuMilli(),
 		MemoryBytes:           req.GetMemoryBytes(),
@@ -658,6 +659,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
 		Spec:                  spec,
 		Scope:                 toAteomSnapshotScope(req.GetScope()),
 		ActorUid:              actorUID,
+		ActorDirs:             actorDirsFor(actorUID),
 	})
 	dAteom = time.Since(tAteom)
 	if err != nil {
@@ -1237,6 +1239,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
 		Spec:                  spec,
 		Scope:                 toAteomSnapshotScope(req.GetScope()),
 		ActorUid:              req.GetActorUid(),
+		ActorDirs:             actorDirsFor(actorUID),
 		EgressGateway:         toAteomEgressGateway(req.GetEgressGateway()),
 		CpuMilli:              req.GetCpuMilli(),
 		MemoryBytes:           req.GetMemoryBytes(),
@@ -1304,6 +1307,7 @@ func (s *AteomHerder) Terminate(ctx context.Context, req *ateletpb.TerminateRequ
 		ActorTemplateName:     req.GetActorTemplateName(),
 		RunscPath:             runscPathFor(assetPaths),
 		Spec:                  spec,
+		ActorDirs:             actorDirsFor(actorUID),
 	}); err != nil {
 		if status.Code(err) == codes.NotFound {
 			slog.InfoContext(ctx, "workload not found on ateom during terminate", slog.Any("actor", actorRef), slog.String("actorUID", actorUID))
@@ -1502,6 +1506,22 @@ func (s *AteomHerder) dialAteom(ctx context.Context, targetAteomUid string) (ate
 	return ateompb.NewAteomClient(conn), nil
 }
 
+// actorDirsFor is the directory set atelet passes to ateom for an actor. ateom
+// takes these from the request rather than deriving them from the actor UID.
+func actorDirsFor(actorUID string) *ateompb.ActorDirs {
+	return &ateompb.ActorDirs{
+		RootDir:                   ateompath.ActorPath(actorUID),
+		OciBundleDir:              ateompath.OCIBundleDir(actorUID),
+		RunscStateDir:             ateompath.RunSCStateDir(actorUID),
+		PidFileDir:                ateompath.PIDFileDir(actorUID),
+		CheckpointDir:             ateompath.CheckpointStateDir(actorUID),
+		RestoreDir:                ateompath.RestoreStateDir(actorUID),
+		DurableDirVolumeMountsDir: ateompath.DurableDirVolumeMountsDir(actorUID),
+		SystemInfoVolumeRootsDir:  ateompath.SystemInfoVolumeRootsDir(actorUID),
+		VolumesDir:                ateompath.VolumesDir(actorUID),
+	}
+}
+
 // buildAteomWorkloadSpec projects the atelet-facing workload spec onto
 // the ateom-facing one.
 func buildAteomWorkloadSpec(spec *ateletpb.WorkloadSpec) (*ateompb.WorkloadSpec, error) {

cmd/atelet/main_test.go

@@ -407,6 +407,35 @@ func TestValidateRestoreRequest(t *testing.T) {
 
 // Every valid atelet scope must map to its ateom counterpart; in particular
 // DATA_ON_GOLDEN must never silently degrade to FULL.
+func TestActorDirsFor(t *testing.T) {
+	const actorUID = "actor-uid-1"
+	dirs := actorDirsFor(actorUID)
+	if dirs.GetRootDir() != ateompath.ActorPath(actorUID) {
+		t.Fatalf("root_dir = %q, want %q", dirs.GetRootDir(), ateompath.ActorPath(actorUID))
+	}
+	under := map[string]string{
+		"oci_bundle_dir":                dirs.GetOciBundleDir(),
+		"runsc_state_dir":               dirs.GetRunscStateDir(),
+		"pid_file_dir":                  dirs.GetPidFileDir(),
+		"checkpoint_dir":                dirs.GetCheckpointDir(),
+		"restore_dir":                   dirs.GetRestoreDir(),
+		"durable_dir_volume_mounts_dir": dirs.GetDurableDirVolumeMountsDir(),
+		"system_info_volume_roots_dir":  dirs.GetSystemInfoVolumeRootsDir(),
+		"volumes_dir":                   dirs.GetVolumesDir(),
+	}
+	seen := map[string]string{}
+	for field, dir := range under {
+		rel, err := filepath.Rel(dirs.GetRootDir(), dir)
+		if err != nil || rel == "." || strings.HasPrefix(rel, "..") {
+			t.Errorf("%s = %q is not a proper subdirectory of root_dir %q", field, dir, dirs.GetRootDir())
+		}
+		if other, dup := seen[dir]; dup {
+			t.Errorf("%s and %s share %q", field, other, dir)
+		}
+		seen[dir] = field
+	}
+}
+
 func TestToAteomSnapshotScope(t *testing.T) {
 	tests := []struct {
 		in   ateletpb.SnapshotScope