DNM: Example of "middle layer" for create.

.gitattributes

@@ -0,0 +1,4 @@
+# Always check-out / check-in files with LF line endings.
+* text=auto eol=lf
+
+**/zz_generated.*.go linguist-generated=true

cmd/ateapi/internal/controlapi/actor.go

@@ -26,33 +26,56 @@ import (
 	"github.com/agent-substrate/substrate/internal/resources"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"github.com/google/uuid"
 	"go.opentelemetry.io/otel/attribute"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/operation"
 	"k8s.io/apimachinery/pkg/api/validate/content"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 )
 
 func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (created *ateapipb.Actor, err error) {
-	if errs := validateCreateActorRequest(req); len(errs) > 0 {
+	// First scrub any fields that users are not allowed to set.
+	inActor := req.Actor
+	if inActor != nil { // otherwise validation will flag it
+		scrubActor(inActor)
+	}
+
+	// Validate the request, including the object within it.
+	if errs := validateCreateActorRequest(ctx, req); len(errs) > 0 {
 		return nil, toGRPCStatusError(errs)
 	}
+
 	start := time.Now()
-	in := req.GetActor()
 	// Recorded only after validation, so every operation uniformly measures a
 	// validated request; malformed ones stay visible in rpc.server.call.duration.
 	defer func() {
 		s.instruments.recordLifecycleOp(ctx, ateattr.OperationCreate, start, err,
-			ateattr.TemplateNameKey.String(in.GetActorTemplateName()),
-			ateattr.TemplateNamespaceKey.String(in.GetActorTemplateNamespace()),
+			ateattr.TemplateNameKey.String(inActor.GetActorTemplateName()),
+			ateattr.TemplateNamespaceKey.String(inActor.GetActorTemplateNamespace()),
 		)
 	}()
-	templateNamespace := in.GetActorTemplateNamespace()
-	templateName := in.GetActorTemplateName()
 
-	setSpanActorRefAttributes(ctx, resources.ActorRefFromActor(in))
+	setSpanActorRefAttributes(ctx, resources.ActorRefFromActor(inActor))
+	// Handle the creation, including validation of the final stored object.
+	stored, err := s.impl.CreateActor(ctx, inActor)
+	setSpanActorAttributes(ctx, stored)
+
+	return stored, err
+}
 
+func (s *ServiceImpl) CreateActor(ctx context.Context, actor *ateapipb.Actor) (*ateapipb.Actor, error) {
+	// Check that the referenced ActorTemplate exists.
+	// FIXME: This is not atomic and it is not a guarantee that the template
+	// will still exist later.  Checking it here produces a nice error UX, but
+	// we still have to handle the template not existing later, which makes the
+	// UX inconsistent, at best.  Is it actually worth checking at all?
+	templateNamespace := actor.GetActorTemplateNamespace()
+	templateName := actor.GetActorTemplateName()
 	template, err := s.actorTemplateLister.ActorTemplates(templateNamespace).Get(templateName)
 	if err != nil {
 		if k8serrors.IsNotFound(err) {
@@ -61,19 +84,24 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
 		return nil, fmt.Errorf("while getting ActorTemplate: %w", err)
 	}
 
+	// If a source snapshot tag is requested, resolve it to a concrete
+	// snapshot.
 	var sourceSnapshotInfo *ateapipb.ActorSnapshotSource
-	if src := in.GetSourceSnapshot(); src != nil {
-		sourceSnapshotInfo, err = s.resolveSnapshotSource(ctx, in.GetMetadata().GetAtespace(), src, template)
+	if src := actor.GetSourceSnapshot(); src != nil {
+		sourceSnapshotInfo, err = s.resolveSnapshotSource(ctx, actor.GetMetadata().GetAtespace(), src, template)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	atespace := in.GetMetadata().GetAtespace()
-	name := in.GetMetadata().GetName()
+	atespace := actor.GetMetadata().GetAtespace()
+	name := actor.GetMetadata().GetName()
 
 	// The atespace must already exist.
-	exists, err := s.persistence.AtespaceExists(ctx, atespace)
+	// FIXME: This is not atomic and it is not a guarantee that the atespace
+	// will still exist when we store the object.  This needs to be part of the
+	// storage operation's contract.
+	exists, err := s.AtespaceExists(ctx, atespace)
 	if err != nil {
 		return nil, fmt.Errorf("while checking atespace: %w", err)
 	}
@@ -87,45 +115,76 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
 		return nil, err
 	}
 
-	actor := &ateapipb.Actor{
-		Metadata: &ateapipb.ResourceMetadata{
-			Atespace: atespace,
-			Name:     name,
-		},
-		Status:                 ateapipb.Actor_STATUS_SUSPENDED,
-		ActorTemplateNamespace: templateNamespace,
-		ActorTemplateName:      templateName,
-		WorkerSelector:         in.GetWorkerSelector(),
-		ActorVolumes:           initVols,
-		LatestSnapshot:         sourceSnapshotInfo.GetSnapshot(),
-		SourceSnapshot:         sourceSnapshotInfo,
-	}
-	stored, err := s.persistence.CreateActor(ctx, actor)
+	// Verify that the result is valid before storing it.
+	outActor := proto.CloneOf(actor)
+	outActor.Status = ateapipb.Actor_STATUS_SUSPENDED
+	outActor.ActorVolumes = initVols
+	outActor.LatestSnapshot = sourceSnapshotInfo.GetSnapshot()
+	outActor.SourceSnapshot = sourceSnapshotInfo
+	if errs := validateActorUpdate(ctx, outActor, actor); len(errs) > 0 {
+		return nil, toGRPCInternalError(errs)
+	}
+
+	// Save the data in the storage layer.
+	stored, err := s.Interface.CreateActor(ctx, outActor)
 	if err != nil {
 		if errors.Is(err, store.ErrAlreadyExists) {
 			return nil, status.Errorf(codes.AlreadyExists, "Actor %s already exists", name)
 		}
-		return nil, fmt.Errorf("while recording actor: %w", err)
+		return nil, fmt.Errorf("while creating actor: %w", err)
 	}
 
-	setSpanActorAttributes(ctx, stored)
 	return stored, nil
 }
 
+// scrubActor removes any fields from the request that clients are not allowed
+// to set.
+func scrubActor(actor *ateapipb.Actor) {
+	// TODO: find a way to do this automatically - proto tags or codegen or something
+	//FIXME: this is obviously wrong for update
+	scrubResourceMetadata(actor.Metadata)
+	actor.Status = 0
+	actor.WorkerAssignment = nil
+	actor.InProgressSnapshotName = ""
+	actor.LatestSnapshot = nil
+	actor.LocalSnapshotInfo = nil
+	actor.InProgressSnapshotSourceActorVersion = 0
+	actor.ActorVolumes = nil
+	actor.InProgressLocalSnapshotName = ""
+	// FIXME: is .SourceSnapshot allowed on input?
+}
+
+// FIXME: put this in a common place for all resources.
+// TODO: find a way to do this automatically - proto tags or codegen or something
+func scrubResourceMetadata(in *ateapipb.ResourceMetadata) {
+	if in == nil {
+		return // validation will flag it
+	}
+	now := timestamppb.Now()
+	*in = ateapipb.ResourceMetadata{
+		Atespace:   in.Atespace,
+		Name:       in.Name,
+		Uid:        uuid.NewString(),
+		Version:    1,
+		CreateTime: now,
+		UpdateTime: now,
+	}
+}
+
 // resolveSnapshotSource resolves a CreateActor request's source snapshot tag
 // and checks that its scope and ActorSnapshot are compatible with creating
 // an Actor in actorAtespace from template.
-func (s *Service) resolveSnapshotSource(ctx context.Context, actorAtespace string, src *ateapipb.ActorSnapshotSource, template *atev1alpha1.ActorTemplate) (*ateapipb.ActorSnapshotSource, error) {
+func (s *ServiceImpl) resolveSnapshotSource(ctx context.Context, actorAtespace string, src *ateapipb.ActorSnapshotSource, template *atev1alpha1.ActorTemplate) (*ateapipb.ActorSnapshotSource, error) {
 	tagRef := src.GetTag()
-	tag, err := s.persistence.GetActorSnapshotTag(ctx, tagRef.GetAtespace(), tagRef.GetName())
+	tag, err := s.GetActorSnapshotTag(ctx, tagRef.GetAtespace(), tagRef.GetName())
 	if errors.Is(err, store.ErrNotFound) {
 		return nil, status.Error(codes.NotFound, "ActorSnapshot not found")
 	}
 	if err != nil {
 		return nil, fmt.Errorf("while getting actor snapshot tag: %w", err)
 	}
 	snapshotRef := tag.GetSnapshot()
-	snapshot, err := s.persistence.GetActorSnapshot(ctx, snapshotRef.GetAtespace(), snapshotRef.GetName())
+	snapshot, err := s.GetActorSnapshot(ctx, snapshotRef.GetAtespace(), snapshotRef.GetName())
 	if errors.Is(err, store.ErrNotFound) {
 		return nil, status.Error(codes.NotFound, "ActorSnapshot not found")
 	}
@@ -161,29 +220,20 @@ func (s *Service) resolveSnapshotSource(ctx context.Context, actorAtespace strin
 	}, nil
 }
 
-func validateCreateActorRequest(req *ateapipb.CreateActorRequest) field.ErrorList {
+func validateCreateActorRequest(ctx context.Context, req *ateapipb.CreateActorRequest) field.ErrorList {
 	var fldPath *field.Path
-	var errs field.ErrorList
+
+	// Call the generated validation.
+	op := operation.Operation{Type: operation.Create, Options: map[string]bool{"validateOutput": false}}
+	errs := Validate_CreateActorRequest(ctx, op, nil, req, nil)
 
 	actor := req.GetActor()
 	actorPath := fldPath.Child("actor")
 	if actor == nil {
-		errs = append(errs, field.Required(actorPath, ""))
+		// handled by DV
 		return errs
 	}
 
-	metaPath := actorPath.Child("metadata")
-	if val, p := actor.GetMetadata().GetAtespace(), metaPath.Child("atespace"); val == "" {
-		errs = append(errs, field.Required(p, ""))
-	} else {
-		errs = append(errs, resources.ValidateResourceName(val, p)...)
-	}
-	if val, p := actor.GetMetadata().GetName(), metaPath.Child("name"); val == "" {
-		errs = append(errs, field.Required(p, ""))
-	} else {
-		errs = append(errs, resources.ValidateResourceName(val, p)...)
-	}
-
 	if val, p := actor.GetActorTemplateNamespace(), actorPath.Child("actor_template_namespace"); val == "" {
 		errs = append(errs, field.Required(p, ""))
 	} else {
@@ -214,7 +264,7 @@ func (s *Service) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (
 		return nil, toGRPCStatusError(errs)
 	}
 	actorRef := resources.ActorRefFromObjectRef(req.GetActor())
-	actor, err := s.persistence.GetActor(ctx, actorRef)
+	actor, err := s.impl.GetActor(ctx, actorRef)
 	if errors.Is(err, store.ErrNotFound) {
 		return nil, status.Errorf(codes.NotFound, "Actor %s not found", actorRef)
 	} else if err != nil {
@@ -241,7 +291,7 @@ func (s *Service) ListActors(ctx context.Context, req *ateapipb.ListActorsReques
 		return nil, toGRPCStatusError(errs)
 	}
 
-	page, err := s.persistence.ListActors(ctx, req.GetAtespace(), store.ListOptions{PageSize: effectivePageSize(req.GetPageSize()), PageToken: req.GetPageToken()})
+	page, err := s.impl.ListActors(ctx, req.GetAtespace(), store.ListOptions{PageSize: effectivePageSize(req.GetPageSize()), PageToken: req.GetPageToken()})
 	if err != nil {
 		return nil, fmt.Errorf("while listing actors in db: %w", err)
 	}
@@ -282,7 +332,7 @@ func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequ
 	actorRef := resources.ActorRefFromActor(in)
 	setSpanActorRefAttributes(ctx, actorRef)
 
-	storedActor, err := s.persistence.UpdateActor(ctx, actorRef, store.WithPrecondition(in, func(toUpdate *ateapipb.Actor) error {
+	storedActor, err := s.impl.UpdateActor(ctx, actorRef, store.WithPrecondition(in, func(toUpdate *ateapipb.Actor) error {
 		fieldmask.Apply(toUpdate, in, req.GetUpdateMask())
 		return nil
 	}))
@@ -468,6 +518,16 @@ func validateSuspendActorRequest(req *ateapipb.SuspendActorRequest) field.ErrorL
 	return errs
 }
 
+func validateActorUpdate(ctx context.Context, newVal, oldVal *ateapipb.Actor) field.ErrorList {
+	var fldPath *field.Path
+
+	// Call the generated validation.
+	op := operation.Operation{Type: operation.Update, Options: map[string]bool{"validateOutput": true}}
+	errs := Validate_Actor(ctx, op, fldPath, newVal, oldVal)
+
+	return errs
+}
+
 func validateSelector(sel *ateapipb.Selector, fldPath *field.Path) field.ErrorList {
 	var errs field.ErrorList
 

cmd/ateapi/internal/controlapi/actor_snapshot.go

@@ -51,7 +51,7 @@ func (s *Service) GetActorSnapshot(ctx context.Context, req *ateapipb.GetActorSn
 	if errs := validateGetActorSnapshotRequest(req); len(errs) > 0 {
 		return nil, toGRPCStatusError(errs)
 	}
-	snapshot, err := s.persistence.GetActorSnapshot(ctx, req.GetSnapshot().GetAtespace(), req.GetSnapshot().GetName())
+	snapshot, err := s.impl.GetActorSnapshot(ctx, req.GetSnapshot().GetAtespace(), req.GetSnapshot().GetName())
 	if errors.Is(err, store.ErrNotFound) {
 		return nil, status.Error(codes.NotFound, "ActorSnapshot not found")
 	}
@@ -78,7 +78,7 @@ func (s *Service) GetActorSnapshotTag(ctx context.Context, req *ateapipb.GetActo
 	if errs := validateGetActorSnapshotTagRequest(req); len(errs) > 0 {
 		return nil, toGRPCStatusError(errs)
 	}
-	tag, err := s.persistence.GetActorSnapshotTag(ctx, req.GetTag().GetAtespace(), req.GetTag().GetName())
+	tag, err := s.impl.GetActorSnapshotTag(ctx, req.GetTag().GetAtespace(), req.GetTag().GetName())
 	if errors.Is(err, store.ErrNotFound) {
 		return nil, status.Error(codes.NotFound, "ActorSnapshot tag not found")
 	}
@@ -105,7 +105,7 @@ func (s *Service) ListActorSnapshots(ctx context.Context, req *ateapipb.ListActo
 	if errs := validateListActorSnapshotsRequest(req); len(errs) > 0 {
 		return nil, toGRPCStatusError(errs)
 	}
-	page, err := s.persistence.ListActorSnapshots(ctx, req.GetAtespace(), store.ListOptions{PageSize: effectivePageSize(req.GetPageSize()), PageToken: req.GetPageToken()})
+	page, err := s.impl.ListActorSnapshots(ctx, req.GetAtespace(), store.ListOptions{PageSize: effectivePageSize(req.GetPageSize()), PageToken: req.GetPageToken()})
 	if err != nil {
 		return nil, fmt.Errorf("while listing actor snapshots: %w", err)
 	}
@@ -136,7 +136,7 @@ func (s *Service) CreateActorSnapshotTag(ctx context.Context, req *ateapipb.Crea
 	if req.GetActorSnapshotTag().GetMetadata().GetAtespace() != ref.GetAtespace() {
 		return nil, status.Error(codes.FailedPrecondition, "ActorSnapshot tags must belong to the snapshot's Atespace")
 	}
-	tag, err := s.persistence.CreateActorSnapshotTag(ctx, ref.GetAtespace(), ref.GetName(), req.GetActorSnapshotTag())
+	tag, err := s.impl.CreateActorSnapshotTag(ctx, ref.GetAtespace(), ref.GetName(), req.GetActorSnapshotTag())
 	if errors.Is(err, store.ErrNotFound) {
 		return nil, status.Error(codes.NotFound, "ActorSnapshot not found")
 	}
@@ -187,7 +187,7 @@ func (s *Service) UpdateActorSnapshotTag(ctx context.Context, req *ateapipb.Upda
 	in := req.GetTag()
 	atespace, name := in.GetMetadata().GetAtespace(), in.GetMetadata().GetName()
 
-	storedTag, err := s.persistence.UpdateActorSnapshotTag(ctx, atespace, name, store.WithPrecondition(in, func(toUpdate *ateapipb.ActorSnapshotTag) error {
+	storedTag, err := s.impl.UpdateActorSnapshotTag(ctx, atespace, name, store.WithPrecondition(in, func(toUpdate *ateapipb.ActorSnapshotTag) error {
 		fieldmask.Apply(toUpdate, in, req.GetUpdateMask())
 		return nil
 	}))
@@ -229,7 +229,7 @@ func (s *Service) DeleteActorSnapshotTag(ctx context.Context, req *ateapipb.Dele
 	if errs := validateDeleteActorSnapshotTagRequest(req); len(errs) > 0 {
 		return nil, toGRPCStatusError(errs)
 	}
-	tag, err := s.persistence.DeleteActorSnapshotTag(ctx, req.GetTag().GetAtespace(), req.GetTag().GetName())
+	tag, err := s.impl.DeleteActorSnapshotTag(ctx, req.GetTag().GetAtespace(), req.GetTag().GetName())
 	if errors.Is(err, store.ErrNotFound) {
 		return nil, status.Errorf(codes.NotFound, "ActorSnapshot tag %s/%s not found", req.GetTag().GetAtespace(), req.GetTag().GetName())
 	}