Bind golden snapshot warmup and publication to Actor revisions

cmd/ateapi/internal/apivalidation/actor_template.go

@@ -58,26 +58,37 @@ func ValidateActorTemplateUpdate(ctx context.Context, fldPath *field.Path, newVa
 
 // ValidateCustom_CreateActorTemplateRequest_ActorTemplate holds the rules
 // that span more than one field of a template: container volume mounts must
-// reference volumes the template declares.
+// reference volumes the template declares, and a VOLUMES preferred fidelity
+// needs a mounted volume snapshots can capture.
 func ValidateCustom_CreateActorTemplateRequest_ActorTemplate(_ context.Context, _ operation.Operation, fldPath *field.Path, value, _ *ateapipb.ActorTemplate) field.ErrorList {
 	var errs field.ErrorList
-	declared := make(map[string]bool, len(value.GetVolumes()))
+	declared := make(map[string]*ateapipb.Volume, len(value.GetVolumes()))
 	for _, vol := range value.GetVolumes() {
-		declared[vol.GetName()] = true
+		declared[vol.GetName()] = vol
 	}
+	hasSnapshotVolume := false
 	for i, ctr := range value.GetContainers() {
 		for j, mount := range ctr.GetVolumeMounts() {
 			name := mount.GetName()
 			if name == "" {
 				continue // required is enforced by tags
 			}
-			if !declared[name] {
+			vol := declared[name]
+			if vol == nil {
 				errs = append(errs, field.Invalid(
 					fldPath.Child("containers").Index(i).Child("volume_mounts").Index(j).Child("name"),
 					name, "must reference a volume declared in the template"))
+				continue
+			}
+			if vol.GetDurableDir() != nil || vol.GetExternalVolumeTemplate() != nil {
+				hasSnapshotVolume = true
 			}
 		}
 	}
+	fidelity := value.GetSnapshotConfig().GetPreferredFidelity()
+	if !hasSnapshotVolume && fidelity == ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES {
+		errs = append(errs, field.Invalid(fldPath.Child("snapshot_config", "preferred_fidelity"), fidelity.String(), "VOLUMES snapshots require a mounted durable-dir or CSI volume"))
+	}
 	return errs
 }
 

cmd/ateapi/internal/apivalidation/actor_template_test.go

@@ -26,6 +26,73 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation/field"
 )
 
+func TestValidateCreateActorTemplateRequestVolumesFidelity(t *testing.T) {
+	// The CSI cases declare external volume templates, which the ExternalVolumes
+	// preview gate controls.
+	setPreviewForTest(t, "ALL")
+	durable := &ateapipb.Volume{Name: "data", DurableDir: &ateapipb.DurableDirVolumeSource{}}
+	csi := &ateapipb.Volume{Name: "csi", ExternalVolumeTemplate: &ateapipb.ExternalVolumeTemplate{
+		Capacity: "1Gi", StorageClassName: "standard",
+	}}
+	image := &ateapipb.Volume{Name: "data", Image: &ateapipb.ImageVolumeSource{
+		Reference: "example.com/data@sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
+	}}
+	systemInfo := &ateapipb.Volume{Name: "system-info", SystemInfo: &ateapipb.SystemInfoVolumeSource{
+		DataSources: []*ateapipb.SystemInfoDataSource{{ActorMetadata: &ateapipb.ActorMetadataDataSource{
+			Items: []*ateapipb.ActorMetadataItem{{Field: ateapipb.ActorMetadataField_ACTOR_METADATA_FIELD_NAME, Path: "actor-name"}},
+		}}},
+	}}
+	for _, snapshots := range []struct {
+		name     string
+		fidelity ateapipb.SnapshotFidelity
+	}{
+		{"memory", ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY},
+		{"volumes", ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES},
+	} {
+		for _, tt := range []struct {
+			name            string
+			volume          *ateapipb.Volume
+			extra           *ateapipb.Volume
+			mounted         bool
+			supportsVolumes bool
+		}{
+			{"no volumes", nil, nil, false, false},
+			{"mounted durable-dir", durable, nil, true, true},
+			{"unmounted durable-dir", durable, nil, false, false},
+			{"mounted CSI", csi, nil, true, true},
+			{"unmounted CSI", csi, nil, false, false},
+			{"mounted image", image, nil, true, false},
+			{"mounted durable-dir and CSI", durable, csi, true, true},
+			{"mounted systemInfo", systemInfo, nil, true, false},
+		} {
+			t.Run(snapshots.name+"/"+tt.name, func(t *testing.T) {
+				tmpl := validActorTemplate(func(tmpl *ateapipb.ActorTemplate) {
+					tmpl.SnapshotConfig.PreferredFidelity = snapshots.fidelity
+					var volumes []*ateapipb.Volume
+					for _, v := range []*ateapipb.Volume{tt.volume, tt.extra} {
+						if v != nil {
+							volumes = append(volumes, v)
+						}
+					}
+					tmpl.Volumes = volumes
+					if tt.mounted {
+						for _, v := range volumes {
+							tmpl.Containers[0].VolumeMounts = append(tmpl.Containers[0].VolumeMounts, &ateapipb.VolumeMount{
+								Name: v.GetName(), MountPath: "/" + v.GetName(),
+							})
+						}
+					}
+				})
+				var want field.ErrorList
+				if !tt.supportsVolumes && snapshots.fidelity == ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES {
+					want = append(want, field.Invalid(field.NewPath("actor_template", "snapshot_config", "preferred_fidelity"), snapshots.fidelity.String(), ""))
+				}
+				assertValidateErr(t, ValidateCreateActorTemplateRequest(context.Background(), &ateapipb.CreateActorTemplateRequest{ActorTemplate: tmpl}), want)
+			})
+		}
+	}
+}
+
 func TestValidateCreateActorTemplateRequest(t *testing.T) {
 	tests := []struct {
 		name string
@@ -67,6 +134,8 @@ func TestValidateCreateActorTemplateRequest(t *testing.T) {
 		"valid volumes fidelity",
 		&ateapipb.CreateActorTemplateRequest{ActorTemplate: validActorTemplate(func(tmpl *ateapipb.ActorTemplate) {
 			tmpl.SnapshotConfig.PreferredFidelity = ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES
+			tmpl.Volumes = []*ateapipb.Volume{{Name: "data", DurableDir: &ateapipb.DurableDirVolumeSource{}}}
+			tmpl.Containers[0].VolumeMounts = []*ateapipb.VolumeMount{{Name: "data", MountPath: "/data"}}
 		})},
 		nil,
 	}, {

cmd/ateapi/internal/controlapi/actor.go

@@ -472,7 +472,7 @@ func (s *RPCService) PauseActor(ctx context.Context, req *ateapipb.PauseActorReq
 	actorRef := resources.ActorRefFromObjectRef(req.GetActor())
 	setSpanActorRefAttributes(ctx, actorRef)
 
-	actor, err := s.actorWorkflow.SuspendActor(ctx, actorRef, suspendAsync)
+	actor, err := s.actorWorkflow.SuspendActor(ctx, actorRef, suspendAsync, nil)
 	if err != nil {
 		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, apierror.Aborted("concurrent update conflict, please retry")
@@ -510,13 +510,17 @@ func (s *RPCService) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorR
 }
 
 func (s *RPCService) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) {
+	return s.suspendActor(ctx, req, nil)
+}
+
+func (s *RPCService) suspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest, precondition *store.Precondition) (*ateapipb.SuspendActorResponse, error) {
 	if errs := apivalidation.ValidateSuspendActorRequest(ctx, req); len(errs) > 0 {
 		return nil, resources.ToAPIError(errs)
 	}
 	actorRef := resources.ActorRefFromObjectRef(req.GetActor())
 	setSpanActorRefAttributes(ctx, actorRef)
 
-	actor, err := s.actorWorkflow.SuspendActor(ctx, actorRef, suspendModeFromOptions(req.GetOptions()))
+	actor, err := s.actorWorkflow.SuspendActor(ctx, actorRef, suspendModeFromOptions(req.GetOptions()), precondition)
 	if err != nil {
 		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, apierror.Aborted("concurrent update conflict, please retry")

cmd/ateapi/internal/controlapi/actor_template_test.go

@@ -172,7 +172,7 @@ func TestDeleteActorTemplate(t *testing.T) {
 				tag = storetest.MustCreateTag(t, ctx, persistence, newPendingTestTag(t, goldenRef.Name, actor))
 			} else {
 				var err error
-				tag, err = workflow.TagActorSnapshot(ctx, tagToCreate(goldenRef, goldenRef.Name))
+				tag, err = workflow.TagActorSnapshot(ctx, tagToCreate(goldenRef, goldenRef.Name), nil)
 				if err != nil {
 					t.Fatal(err)
 				}

cmd/ateapi/internal/controlapi/export_test.go

@@ -87,7 +87,7 @@ func ReleaseWindowWorkflows() []ReleaseWindowWorkflow {
 			// and its first actor update is the finalize commit.
 			Finalize: func(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error {
 				w, _ := newWireCaptureWorkflow(t, st)
-				_, err := w.SuspendActor(ctx, actorRef, suspendAsync)
+				_, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil)
 				return err
 			},
 			WantRetry:      codes.OK,
@@ -102,7 +102,7 @@ func ReleaseWindowWorkflows() []ReleaseWindowWorkflow {
 			// As for the pause, the finalize commit is the first actor update.
 			Finalize: func(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, _ *ateapipb.Actor) error {
 				w, _ := newWireCaptureWorkflow(t, st)
-				_, err := w.SuspendActor(ctx, actorRef, suspendSync)
+				_, err := w.SuspendActor(ctx, actorRef, suspendSync, nil)
 				return err
 			},
 			WantRetry:      codes.OK,

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

@@ -3259,10 +3259,12 @@ func createDataFidelityTemplate(t *testing.T, tc *testContext, ns string) *ateap
 				PreferredFidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES,
 			},
 			Containers: []*ateapipb.Container{{
-				Name:    "main",
-				Image:   "main@sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
-				Command: []string{"/main"},
+				Name:         "main",
+				Image:        "main@sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
+				Command:      []string{"/main"},
+				VolumeMounts: []*ateapipb.VolumeMount{{Name: "data", MountPath: "/data"}},
 			}},
+			Volumes: []*ateapipb.Volume{{Name: "data", DurableDir: &ateapipb.DurableDirVolumeSource{}}},
 			WorkerSelector: &ateapipb.Selector{
 				MatchLabels: map[string]string{poolLabelKey: ns},
 			},

cmd/ateapi/internal/controlapi/tag.go

@@ -34,6 +34,10 @@ import (
 // again or deleted cannot collect it. The work is a workflow because it spans
 // two transactions around an object copy; see TagActorSnapshot.
 func (s *RPCService) CreateTag(ctx context.Context, req *ateapipb.CreateTagRequest) (*ateapipb.Tag, error) {
+	return s.createTag(ctx, req, nil)
+}
+
+func (s *RPCService) createTag(ctx context.Context, req *ateapipb.CreateTagRequest, precondition *store.Precondition) (*ateapipb.Tag, error) {
 	// First scrub any fields that users are not allowed to set, then fill the
 	// defaults so validation sees the final resource state.
 	inTag := req.Tag
@@ -49,7 +53,7 @@ func (s *RPCService) CreateTag(ctx context.Context, req *ateapipb.CreateTagReque
 	actorRef := resources.ActorRefFromObjectRef(req.GetTag().GetSourceActor())
 	setSpanActorRefAttributes(ctx, actorRef)
 
-	tag, err := s.actorWorkflow.TagActorSnapshot(ctx, req.GetTag())
+	tag, err := s.actorWorkflow.TagActorSnapshot(ctx, req.GetTag(), precondition)
 	if err != nil {
 		if errors.Is(err, store.ErrNotFound) {
 			return nil, apierror.FailedPrecondition("Actor %s not found", actorRef)