From 491cd80b0b195b02180e43cb06c8fe993070f03d Mon Sep 17 00:00:00 2001 From: cloudonly Date: Mon, 12 Oct 2026 10:47:15 +0800 Subject: [PATCH 1/2] ateapi: reject VOLUMES snapshots without supported volume mounts A template whose preferred fidelity is VOLUMES but that mounts no volume a snapshot can capture (durable-dir, or a CSI volume on the micro-VM class) fails at runtime on every pause and suspend. Reject it at template admission instead, in the custom validation that already checks volume references and the ROOTFS sandbox class. Document the requirement next to the ROOTFS rule in the API guide, including that an external volume's data stays on the volume rather than in the snapshot. --- .../internal/apivalidation/actor_template.go | 19 +++-- .../apivalidation/actor_template_test.go | 69 +++++++++++++++++++ .../controlapi/functionaltest/actor_test.go | 8 ++- docs/api-guide.md | 2 +- pkg/proto/ateapipb/ateapi.pb.go | 2 +- pkg/proto/ateapipb/ateapi.proto | 2 +- 6 files changed, 92 insertions(+), 10 deletions(-) diff --git a/cmd/ateapi/internal/apivalidation/actor_template.go b/cmd/ateapi/internal/apivalidation/actor_template.go index 2b558e74b2..f670f0782b 100644 --- a/cmd/ateapi/internal/apivalidation/actor_template.go +++ b/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 } diff --git a/cmd/ateapi/internal/apivalidation/actor_template_test.go b/cmd/ateapi/internal/apivalidation/actor_template_test.go index 7b4d963cee..205962cd93 100644 --- a/cmd/ateapi/internal/apivalidation/actor_template_test.go +++ b/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, }, { diff --git a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go index bf4b207465..eb3c0f92d6 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go +++ b/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}, }, diff --git a/docs/api-guide.md b/docs/api-guide.md index d555516f09..25b5f97877 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -100,7 +100,7 @@ The `ActorTemplate` defines the code, environment, and state-management policies | :--- | :--- | :--- | | `containers` | `[]Container` | **Required.** The workload definition — see [Container Fields](#container-fields) below. Each container may also declare an optional `wakeupProbe` HTTP probe — see [Container Wakeup Probe](#container-wakeup-probe-wakeupprobe). | | `workerSelector` | `*Selector` | Optional. Gates which `WorkerPool`s actors from this template may use, by matching against each pool's labels (`matchLabels`). If unset, all pools are eligible (subject to the actor's own `worker_selector`). The pool decides the sandbox class, so this is also how a template pins one: if your cluster has pools of more than one sandbox class and the workload needs a particular class, make sure the selector points only at pools of that class. An actor booting with no snapshot and no selector may land on any class, and keeps it from its first checkpoint on. | -| `snapshotConfig` | `SnapshotConfig` | **Required.** The base object-storage location snapshots are written under, the snapshot fidelity (`preferredFidelity`) every Pause and Suspend captures (`SNAPSHOT_FIDELITY_ROOTFS` is supported on `microvm` pools), and automated golden snapshot creation (`goldenSnapshotConfig.mode`). See [Snapshot Storage Layout](#snapshot-storage-layout). | +| `snapshotConfig` | `SnapshotConfig` | **Required.** The base object-storage location snapshots are written under, the snapshot fidelity (`preferredFidelity`) every Pause and Suspend captures (`SNAPSHOT_FIDELITY_ROOTFS` is supported on `microvm` pools; `SNAPSHOT_FIDELITY_VOLUMES` requires a container to mount a `durableDir` volume, or an `externalVolumeTemplate` volume whose data stays on the volume rather than in the snapshot), and automated golden snapshot creation (`goldenSnapshotConfig.mode`). See [Snapshot Storage Layout](#snapshot-storage-layout). | | `volumes` | `[]Volume` | Optional. Volumes the containers may mount, each a `durableDir`, an `externalVolumeTemplate` (see [CSI Volumes Guide](csi-volumes.md)), or a `systemInfo` volume (see [SystemInfo Volumes](#systeminfo-volumes)). Every declared volume must be mounted by at least one container. A `microvm` worker supports several `durableDir` volumes; a `gvisor` worker is limited to one. A `durableDir` is created `0777`, as Kubernetes creates an emptyDir, so a container running as any user can write it. A snapshot restores it with the mode it had, so the container's own chmod survives suspend and resume. | | `resources` | `*ResourceRequirements` | Required. Declares each actor's compute size via `limits`, which must set both `cpu` and `memory` — see [Sandbox Right-Sizing](#sandbox-right-sizing-resources). Immutable, like the rest of the template. | diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index 84bb8d812d..56f840facb 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -4459,7 +4459,7 @@ type CreateActorTemplateRequest struct { // version, timestamps) is ignored, as are the status fields. // // +k8s:required - // +k8s:customValidation # volume_mounts must reference declared volumes + // +k8s:customValidation # volume_mounts must reference declared volumes; VOLUMES fidelity requires a snapshot-capable mounted volume ActorTemplate *ActorTemplate `protobuf:"bytes,1,opt,name=actor_template,json=actorTemplate,proto3" json:"actor_template,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 691f130cbc..bbb5e5e3c0 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -1661,7 +1661,7 @@ message CreateActorTemplateRequest { // version, timestamps) is ignored, as are the status fields. // // +k8s:required - // +k8s:customValidation # volume_mounts must reference declared volumes + // +k8s:customValidation # volume_mounts must reference declared volumes; VOLUMES fidelity requires a snapshot-capable mounted volume ActorTemplate actor_template = 1; } From c85a6c08865bb59e2909ca50a7d9de7a58cdab80 Mon Sep 17 00:00:00 2001 From: cloudonly Date: Tue, 13 Oct 2026 21:33:46 +0800 Subject: [PATCH 2/2] Bind golden snapshot warmup and publication to Actor revisions Warmup could outlive the Actor revision it started on, and recovery assumed that an existing durable snapshot had been approved for publication. Bind the warmup timer to the observed RUNNING Actor's UID and version, and publish only a capture completed during the current reconciliation attempt. Suspend checks the warmed revision under the actor lease. CreateTag checks the revision Suspend returned under its own lease, so an intervening lifecycle operation aborts publication rather than tagging an unapproved snapshot. Approval lives only in the reconciler's memory: an interrupted capture or failed copy resumes the existing guest state and starts another warmup, while a completed tag is recovered before retrying cleanup. Adds the server-owned warmup_actor_uid and warmup_actor_version fields to GoldenSnapshotStatus. --- .../apivalidation/zz_generated.validation.go | 62 +++++ cmd/ateapi/internal/controlapi/actor.go | 8 +- .../controlapi/actor_template_test.go | 2 +- cmd/ateapi/internal/controlapi/export_test.go | 4 +- cmd/ateapi/internal/controlapi/tag.go | 6 +- .../controlapi/template_consistency_test.go | 258 ++++++++++++++++++ .../controlapi/template_reconciler.go | 75 ++--- .../controlapi/template_reconciler_test.go | 158 ++++++++--- .../controlapi/workflow_delete_test.go | 2 +- .../controlapi/workflow_pause_test.go | 26 +- .../controlapi/workflow_revert_test.go | 2 +- .../internal/controlapi/workflow_suspend.go | 14 +- .../controlapi/workflow_suspend_test.go | 84 ++++-- .../internal/controlapi/workflow_tag.go | 13 +- .../internal/controlapi/workflow_tag_test.go | 65 ++++- .../internal/controlapi/workflow_test.go | 4 +- pkg/proto/ateapipb/ateapi.pb.go | 40 ++- pkg/proto/ateapipb/ateapi.proto | 16 +- 18 files changed, 693 insertions(+), 146 deletions(-) create mode 100644 cmd/ateapi/internal/controlapi/template_consistency_test.go diff --git a/cmd/ateapi/internal/apivalidation/zz_generated.validation.go b/cmd/ateapi/internal/apivalidation/zz_generated.validation.go index 9c16957d28..25b9ec582d 100644 --- a/cmd/ateapi/internal/apivalidation/zz_generated.validation.go +++ b/cmd/ateapi/internal/apivalidation/zz_generated.validation.go @@ -4625,6 +4625,68 @@ func Validate_GoldenSnapshotStatus( errs = append(errs, fn(fldPath.Child("error_message"), &obj.ErrorMessage, oldVal, oldObj != nil)...) } + { // field ateapipb.GoldenSnapshotStatus.WarmupActorUid + fn := func( + fldPath *field.Path, + obj, oldObj *string, + oldValueCorrelated bool) (errs field.ErrorList) { + // don't revalidate unchanged data + if oldValueCorrelated && op.Type == operation.Update { + if obj == oldObj || (obj != nil && oldObj != nil && *obj == *oldObj) { + return nil + } + } + // call field-attached validations + earlyReturn := false + if e := validate.OptionalValue(ctx, op, fldPath, obj, oldObj).MarkShortCircuit(); len(e) != 0 { + earlyReturn = true + } + if earlyReturn { + return // do not proceed + } + if e := validate.UUID(ctx, op, fldPath, obj, oldObj); len(e) != 0 { + errs = append(errs, e...) + } + return + } + oldVal := safe.Field(oldObj, + func(oldObj *ateapipb.GoldenSnapshotStatus) *string { + return &oldObj.WarmupActorUid + }) + errs = append(errs, fn(fldPath.Child("warmup_actor_uid"), &obj.WarmupActorUid, oldVal, oldObj != nil)...) + } + + { // field ateapipb.GoldenSnapshotStatus.WarmupActorVersion + fn := func( + fldPath *field.Path, + obj, oldObj *int64, + oldValueCorrelated bool) (errs field.ErrorList) { + // don't revalidate unchanged data + if oldValueCorrelated && op.Type == operation.Update { + if obj == oldObj || (obj != nil && oldObj != nil && *obj == *oldObj) { + return nil + } + } + // call field-attached validations + earlyReturn := false + if e := validate.OptionalValue(ctx, op, fldPath, obj, oldObj).MarkShortCircuit(); len(e) != 0 { + earlyReturn = true + } + if earlyReturn { + return // do not proceed + } + if e := validate.Minimum(ctx, op, fldPath, obj, oldObj, 1); len(e) != 0 { + errs = append(errs, e...) + } + return + } + oldVal := safe.Field(oldObj, + func(oldObj *ateapipb.GoldenSnapshotStatus) *int64 { + return &oldObj.WarmupActorVersion + }) + errs = append(errs, fn(fldPath.Child("warmup_actor_version"), &obj.WarmupActorVersion, oldVal, oldObj != nil)...) + } + return errs } diff --git a/cmd/ateapi/internal/controlapi/actor.go b/cmd/ateapi/internal/controlapi/actor.go index c25a36f775..169cbc643a 100644 --- a/cmd/ateapi/internal/controlapi/actor.go +++ b/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") diff --git a/cmd/ateapi/internal/controlapi/actor_template_test.go b/cmd/ateapi/internal/controlapi/actor_template_test.go index 501b1671f4..6ef6ccc634 100644 --- a/cmd/ateapi/internal/controlapi/actor_template_test.go +++ b/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) } diff --git a/cmd/ateapi/internal/controlapi/export_test.go b/cmd/ateapi/internal/controlapi/export_test.go index 258befa692..9b87a287e9 100644 --- a/cmd/ateapi/internal/controlapi/export_test.go +++ b/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, diff --git a/cmd/ateapi/internal/controlapi/tag.go b/cmd/ateapi/internal/controlapi/tag.go index 4bff1c88f3..5631ca8c31 100644 --- a/cmd/ateapi/internal/controlapi/tag.go +++ b/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) diff --git a/cmd/ateapi/internal/controlapi/template_consistency_test.go b/cmd/ateapi/internal/controlapi/template_consistency_test.go new file mode 100644 index 0000000000..7781e836fd --- /dev/null +++ b/cmd/ateapi/internal/controlapi/template_consistency_test.go @@ -0,0 +1,258 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controlapi + +import ( + "context" + "errors" + "testing" + "testing/synctest" + "time" + + "go.ate.dev/substrate/cmd/ateapi/internal/store" + "go.ate.dev/substrate/internal/apierror" + "go.ate.dev/substrate/internal/resources" + "go.ate.dev/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" +) + +func TestReconcileOne_InterruptedCaptureRestartsWarmup(t *testing.T) { + for _, state := range []ateapipb.ActorState{ + ateapipb.ActorState_ACTOR_STATE_SUSPENDING, + ateapipb.ActorState_ACTOR_STATE_SUSPENDED, + } { + t.Run(state.String(), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + st := newFakeTemplateStore(testTemplate(withoutWakeupProbe)) + control := &fakeGoldenControl{exists: true, goldenState: state, goldenSnapshot: "gs://bucket/interrupted"} + r := newTestTemplateReconciler(st, control) + defer r.queue.ShutDown() + delay, err := r.reconcileOne(t.Context(), testTemplateRef) + if err != nil { + t.Fatal(err) + } + if delay != goldenSnapshotWarmup || st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetGoldenTag() != nil { + t.Fatalf("interrupted capture returned delay %s and tag %v, want a fresh warmup", delay, st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetGoldenTag()) + } + }) + }) + } +} + +type interferingGoldenControl struct { + *fakeGoldenControl + beforeCapture bool + beforePublication bool +} + +func (c *interferingGoldenControl) suspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest, precondition *store.Precondition) (*ateapipb.SuspendActorResponse, error) { + if c.beforeCapture { + c.beforeCapture = false + if _, err := c.fakeGoldenControl.PauseActor(ctx, &ateapipb.PauseActorRequest{Actor: req.GetActor()}); err != nil { + return nil, err + } + } + return c.fakeGoldenControl.suspendActor(ctx, req, precondition) +} + +func (c *interferingGoldenControl) createTag(ctx context.Context, req *ateapipb.CreateTagRequest, precondition *store.Precondition) (*ateapipb.Tag, error) { + if c.beforePublication { + c.beforePublication = false + ref := req.GetTag().GetSourceActor() + if _, err := c.fakeGoldenControl.ResumeActor(ctx, &ateapipb.ResumeActorRequest{Actor: ref}); err != nil { + return nil, err + } + if _, err := c.fakeGoldenControl.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: ref}); err != nil { + return nil, err + } + } + return c.fakeGoldenControl.createTag(ctx, req, precondition) +} + +func TestReconcileOne_CapturePublicationRacesAreRejected(t *testing.T) { + for _, phase := range []string{"capture", "publication"} { + t.Run(phase, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + st := newFakeTemplateStore(testTemplate(withoutWakeupProbe)) + control := &interferingGoldenControl{ + fakeGoldenControl: &fakeGoldenControl{snapshot: "gs://bucket/snapshot"}, + beforeCapture: phase == "capture", + beforePublication: phase == "publication", + } + r := newTestTemplateReconciler(st, control) + defer r.queue.ShutDown() + if _, err := r.reconcileOne(t.Context(), testTemplateRef); err != nil { + t.Fatal(err) + } + time.Sleep(goldenSnapshotWarmup) + if _, err := r.reconcileOne(t.Context(), testTemplateRef); apierror.Code(err) != codes.Aborted { + t.Fatalf("%s race returned %v, want Aborted", phase, err) + } + if st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetGoldenTag() != nil { + t.Fatal("published a snapshot after a conflicting lifecycle operation") + } + }) + }) + } +} + +func TestReconcileOne_HiddenLifecycleRestartsWarmup(t *testing.T) { + for _, operation := range []string{"pause", "suspend", "revert", "replace"} { + t.Run(operation, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + st := newFakeTemplateStore(testTemplate(withoutWakeupProbe)) + control := &fakeGoldenControl{snapshot: "gs://bucket/snapshot"} + r := newTestTemplateReconciler(st, control) + defer r.queue.ShutDown() + if _, err := r.reconcileOne(t.Context(), testTemplateRef); err != nil { + t.Fatal(err) + } + ref := &ateapipb.ObjectRef{Atespace: resources.GoldenActorAtespace, Name: testTemplateUID} + time.Sleep(5 * time.Second) + var err error + switch operation { + case "pause": + _, err = control.PauseActor(t.Context(), &ateapipb.PauseActorRequest{Actor: ref}) + case "suspend": + _, err = control.SuspendActor(t.Context(), &ateapipb.SuspendActorRequest{Actor: ref}) + case "revert": + _, err = control.RevertActor(t.Context(), &ateapipb.RevertActorRequest{Actor: ref}) + case "replace": + control.actorUID = "22222222-2222-4222-8222-222222222222" + } + if err != nil { + t.Fatal(err) + } + time.Sleep(20 * time.Second) + if _, err := control.ResumeActor(t.Context(), &ateapipb.ResumeActorRequest{Actor: ref}); err != nil { + t.Fatal(err) + } + delay, err := r.reconcileOne(t.Context(), testTemplateRef) + if err != nil { + t.Fatal(err) + } + if delay != goldenSnapshotWarmup || st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetGoldenTag() != nil { + t.Fatalf("%s inherited old warmup: delay %s, tag %v", operation, delay, st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetGoldenTag()) + } + }) + }) + } +} + +type slowGoldenResume struct { + *fakeGoldenControl + resumedAt time.Time + taggedAt time.Time +} + +func (c *slowGoldenResume) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequest) (*ateapipb.ResumeActorResponse, error) { + time.Sleep(25 * time.Second) + resp, err := c.fakeGoldenControl.ResumeActor(ctx, req) + c.resumedAt = time.Now() + return resp, err +} + +func (c *slowGoldenResume) createTag(ctx context.Context, req *ateapipb.CreateTagRequest, precondition *store.Precondition) (*ateapipb.Tag, error) { + c.taggedAt = time.Now() + return c.fakeGoldenControl.createTag(ctx, req, precondition) +} + +func TestReconcileOne_WarmupAfterSlowResume(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + st := newFakeTemplateStore(testTemplate(withoutWakeupProbe)) + control := &slowGoldenResume{fakeGoldenControl: &fakeGoldenControl{snapshot: "gs://bucket/snapshot"}} + r := newTestTemplateReconciler(st, control) + defer r.queue.ShutDown() + delay, err := r.reconcileOne(t.Context(), testTemplateRef) + if err != nil { + t.Fatal(err) + } + if !control.taggedAt.IsZero() { + t.Fatalf("snapshot tagged %s after Resume returned; want a full %s warmup", control.taggedAt.Sub(control.resumedAt), goldenSnapshotWarmup) + } + if delay < goldenSnapshotWarmup { + t.Fatalf("remaining warmup = %s, want >= %s", delay, goldenSnapshotWarmup) + } + time.Sleep(delay) + if _, err := r.reconcileOne(t.Context(), testTemplateRef); err != nil { + t.Fatal(err) + } + if control.taggedAt.Sub(control.resumedAt) < goldenSnapshotWarmup { + t.Fatal("resume time was counted as warmup") + } + }) +} + +func TestReconcileOne_PublicationRetry(t *testing.T) { + for _, failure := range []string{"copy", "cleanup"} { + t.Run(failure, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + st := newFakeTemplateStore(testTemplate(withoutWakeupProbe)) + control := &fakeGoldenControl{snapshot: "gs://bucket/snapshot"} + r := newTestTemplateReconciler(st, control) + defer r.queue.ShutDown() + if _, err := r.reconcileOne(t.Context(), testTemplateRef); err != nil { + t.Fatal(err) + } + time.Sleep(goldenSnapshotWarmup) + if failure == "copy" { + control.tagErr = errors.New("copy failed") + } else { + control.deleteErr = errors.New("cleanup failed") + } + if _, err := r.reconcileOne(t.Context(), testTemplateRef); err == nil { + t.Fatal("injected failure succeeded") + } + control.tagErr, control.deleteErr = nil, nil + delay, err := r.reconcileOne(t.Context(), testTemplateRef) + if err != nil { + t.Fatal(err) + } + wantCalls := 1 + if failure == "copy" { + wantCalls = 2 + if delay != goldenSnapshotWarmup || st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetGoldenTag() != nil { + t.Fatal("failed copy reused approval from a previous pass") + } + time.Sleep(delay) + if _, err := r.reconcileOne(t.Context(), testTemplateRef); err != nil { + t.Fatal(err) + } + } else if delay != 0 { + t.Fatal("completed tag restarted warmup") + } + if st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetGoldenTag() == nil { + t.Fatal("retry did not publish the tag") + } + if _, resumes, suspends := control.callCounts(); resumes != wantCalls || suspends != wantCalls { + t.Fatalf("retry resumed %d and suspended %d times, want %d each", resumes, suspends, wantCalls) + } + }) + }) + } +} + +func TestReconcileOne_ChangedTemplateFailsBuild(t *testing.T) { + st := newFakeTemplateStore(testTemplate()) + control := &fakeGoldenControl{exists: true, goldenState: ateapipb.ActorState_ACTOR_STATE_RUNNING, actorTemplate: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "other-template"}} + r := newTestTemplateReconciler(st, control) + defer r.queue.ShutDown() + if _, err := r.reconcileOne(t.Context(), testTemplateRef); err != nil { + t.Fatal(err) + } + if st.storedStatus(t, testTemplateRef).GetGoldenSnapshotStatus().GetErrorMessage() == "" || len(control.suspendReqs) != 0 || len(control.tagReqs) != 0 { + t.Fatal("captured or published an actor that uses another template") + } +} diff --git a/cmd/ateapi/internal/controlapi/template_reconciler.go b/cmd/ateapi/internal/controlapi/template_reconciler.go index cbb583de08..4b3054c001 100644 --- a/cmd/ateapi/internal/controlapi/template_reconciler.go +++ b/cmd/ateapi/internal/controlapi/template_reconciler.go @@ -69,14 +69,14 @@ type templateReconcilerStore interface { // reconciler drives golden actors through. *RPCService satisfies it. type goldenActorControl interface { GetTag(ctx context.Context, req *ateapipb.GetTagRequest) (*ateapipb.Tag, error) - CreateTag(ctx context.Context, req *ateapipb.CreateTagRequest) (*ateapipb.Tag, error) + createTag(ctx context.Context, req *ateapipb.CreateTagRequest, precondition *store.Precondition) (*ateapipb.Tag, error) DeleteTag(ctx context.Context, req *ateapipb.DeleteTagRequest) (*ateapipb.Tag, error) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequest) (*ateapipb.Actor, error) CreateAtespace(ctx context.Context, req *ateapipb.CreateAtespaceRequest) (*ateapipb.Atespace, error) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (*ateapipb.Actor, error) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (*ateapipb.Actor, error) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequest) (*ateapipb.ResumeActorResponse, error) - SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) + suspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest, precondition *store.Precondition) (*ateapipb.SuspendActorResponse, error) } // ActorTemplateReconciler drives stored ActorTemplates through the golden @@ -248,6 +248,9 @@ func (r *ActorTemplateReconciler) reconcileOne(ctx context.Context, ref resource } return 0, err } + if resources.ActorTemplateRefFromObjectRef(actor.GetActorTemplate()) != ref { + return 0, r.fail(ctx, tmpl, reasonUnexpectedState, "golden actor no longer uses this template") + } switch state := actor.GetStatus().GetState(); state { case ateapipb.ActorState_ACTOR_STATE_CRASHED: @@ -255,14 +258,22 @@ func (r *ActorTemplateReconciler) reconcileOne(ctx context.Context, ref resource case ateapipb.ActorState_ACTOR_STATE_RUNNING: takeAt := goldenSnapshotStatus.GetSnapshotTime() - if takeAt == nil { - // Resumed, but the deadline write was lost (a replica died - // after ResumeActor). The elapsed warmup is unknowable, so - // restart the clock. - slog.WarnContext(ctx, "Golden actor running without a snapshot deadline; restarting warmup", slog.String("ActorTemplate", ref.String())) + if takeAt == nil || + goldenSnapshotStatus.GetWarmupActorUid() != actor.GetMetadata().GetUid() || + goldenSnapshotStatus.GetWarmupActorVersion() != actor.GetMetadata().GetVersion() { + // A new incarnation or lifecycle revision cannot inherit elapsed warmup. + if takeAt != nil || goldenSnapshotStatus.GetWarmupActorUid() != "" { + // Interference restarted the clock; without this line a build + // that never finishes leaves no trace of why. + slog.InfoContext(ctx, "Golden actor changed revision; restarting warmup", + slog.String("ActorTemplate", ref.String()), + slog.String("Actor", actor.GetMetadata().GetUid())) + } deadline := time.Now().Add(goldenSnapshotWarmupFor(tmpl.GetContainers())) if tmpl, err = r.checkpoint(ctx, tmpl, func(snapshotStatus *ateapipb.GoldenSnapshotStatus) { snapshotStatus.SnapshotTime = timestamppb.New(deadline) + snapshotStatus.WarmupActorUid = actor.GetMetadata().GetUid() + snapshotStatus.WarmupActorVersion = actor.GetMetadata().GetVersion() }); err != nil { return 0, err } @@ -273,40 +284,36 @@ func (r *ActorTemplateReconciler) reconcileOne(ctx context.Context, ref resource return rem, nil } // Warmup done: suspend the golden actor and record its snapshot. - err := r.suspendActor(ctx, goldenActorRef) + captured, err := r.suspendActor(ctx, goldenActorRef, actor) if err != nil { return 0, err } - return 0, r.tagGoldenActor(ctx, tmpl, goldenActorRef) + return 0, r.tagGoldenActor(ctx, tmpl, goldenActorRef, captured) case ateapipb.ActorState_ACTOR_STATE_SUSPENDING: - // A previous pass died mid-suspend; retry suspend. - err := r.suspendActor(ctx, goldenActorRef) + // Finish an interrupted suspend, then resume for a fresh warmup. + // Approval to publish does not survive the reconcile pass. + _, err := r.suspendActor(ctx, goldenActorRef, actor) if err != nil { return 0, err } - return 0, r.tagGoldenActor(ctx, tmpl, goldenActorRef) + continue case ateapipb.ActorState_ACTOR_STATE_RESUMING, ateapipb.ActorState_ACTOR_STATE_SUSPENDED: - // The golden actor was never resumed, or a previous resume didn't - // finish; ResumeActor is reentrant from both. - if findLatestDurableSnapshot(actor.GetStatus(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED) != nil { - // Golden actors never start from a source snapshot, so an - // existing snapshot means an earlier suspend completed - // without being recorded. - return 0, r.tagGoldenActor(ctx, tmpl, goldenActorRef) + // Forget the previous timer before resuming. Only a RUNNING observation + // can start a new one; guest state is retained when the Actor resumes. + if tmpl, err = r.checkpoint(ctx, tmpl, func(snapshotStatus *ateapipb.GoldenSnapshotStatus) { + snapshotStatus.SnapshotTime = nil + snapshotStatus.WarmupActorUid = "" + snapshotStatus.WarmupActorVersion = 0 + }); err != nil { + return 0, err } if _, err := r.control.ResumeActor(ctx, &ateapipb.ResumeActorRequest{Actor: goldenActorRef}); err != nil { // A crash during resume is observed as CRASHED on the retry. return 0, fmt.Errorf("while resuming golden actor: %w", err) } - deadline := time.Now().Add(goldenSnapshotWarmupFor(tmpl.GetContainers())) - if tmpl, err = r.checkpoint(ctx, tmpl, func(snapshotStatus *ateapipb.GoldenSnapshotStatus) { - snapshotStatus.SnapshotTime = timestamppb.New(deadline) - }); err != nil { - return 0, err - } case ateapipb.ActorState_ACTOR_STATE_DELETING, ateapipb.ActorState_ACTOR_STATE_REVERTING: // Nothing in the golden flow deletes or reverts the actor before // the snapshot is taken; someone else interfered. @@ -320,25 +327,27 @@ func (r *ActorTemplateReconciler) reconcileOne(ctx context.Context, ref resource // suspendActor waits for the golden actor to produce an external snapshot. // SuspendActor completes an in-flight suspend and is a no-op if already suspended. -func (r *ActorTemplateReconciler) suspendActor(ctx context.Context, goldenRef *ateapipb.ObjectRef) error { - resp, err := r.control.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: goldenRef}) +func (r *ActorTemplateReconciler) suspendActor(ctx context.Context, goldenRef *ateapipb.ObjectRef, observed *ateapipb.Actor) (*ateapipb.Actor, error) { + precondition := store.PreconditionFrom(observed) + resp, err := r.control.suspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: goldenRef}, &precondition) if err != nil { // A crash during suspend is observed as CRASHED on the retry. - return fmt.Errorf("while suspending golden actor: %w", err) + return nil, fmt.Errorf("while suspending golden actor: %w", err) } if findLatestDurableSnapshot(resp.GetActor().GetStatus(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED) == nil { - return fmt.Errorf("suspending golden actor produced no external snapshot") + return nil, fmt.Errorf("suspending golden actor produced no external snapshot") } - return nil + return resp.GetActor(), nil } // tagGoldenActor copies the snapshot into a tag before releasing the actor's copy. -func (r *ActorTemplateReconciler) tagGoldenActor(ctx context.Context, tmpl *ateapipb.ActorTemplate, ref *ateapipb.ObjectRef) error { - _, err := r.control.CreateTag(ctx, &ateapipb.CreateTagRequest{Tag: &ateapipb.Tag{ +func (r *ActorTemplateReconciler) tagGoldenActor(ctx context.Context, tmpl *ateapipb.ActorTemplate, ref *ateapipb.ObjectRef, captured *ateapipb.Actor) error { + precondition := store.PreconditionFrom(captured) + _, err := r.control.createTag(ctx, &ateapipb.CreateTagRequest{Tag: &ateapipb.Tag{ Metadata: &ateapipb.ResourceMetadata{Atespace: ref.GetAtespace(), Name: ref.GetName()}, SourceActor: ref, Scope: ateapipb.TagScope_TAG_SCOPE_PUBLISHED, - }}) + }}, &precondition) if err != nil { return fmt.Errorf("while creating golden tag: %w", err) } diff --git a/cmd/ateapi/internal/controlapi/template_reconciler_test.go b/cmd/ateapi/internal/controlapi/template_reconciler_test.go index 274aae5a45..b931fc1add 100644 --- a/cmd/ateapi/internal/controlapi/template_reconciler_test.go +++ b/cmd/ateapi/internal/controlapi/template_reconciler_test.go @@ -146,6 +146,7 @@ func (s *fakeTemplateStore) storedStatus(t *testing.T, ref resources.ActorTempla // goldenState / goldenSnapshot. type fakeGoldenControl struct { mu sync.Mutex + pausedLocal bool tag *ateapipb.Tag tagErr error deleteErr error @@ -157,10 +158,16 @@ type fakeGoldenControl struct { resumeErr error suspendErr error getErr error + pauseErr error // exists seeds whether the golden actor pre-exists; goldenState and // goldenSnapshot are its observed state and external snapshot while it // does. + actorUID string + actorTemplate *ateapipb.ObjectRef + generation int32 + version int64 + pauseReqs []*ateapipb.PauseActorRequest exists bool goldenState ateapipb.ActorState goldenSnapshot string @@ -181,13 +188,48 @@ func (c *fakeGoldenControl) CreateAtespace(_ context.Context, req *ateapipb.Crea return req.GetAtespace(), nil } +func (c *fakeGoldenControl) goldenActorUID() string { + if c.actorUID != "" { + return c.actorUID + } + return someActorUID +} + +func (c *fakeGoldenControl) goldenActorVersion() int64 { + if c.version == 0 { + return 1 + } + return c.version +} + +func (c *fakeGoldenControl) goldenActorTemplate() *ateapipb.ObjectRef { + if c.actorTemplate != nil { + return c.actorTemplate + } + return testTemplateRef.ToObjectRef() +} + func (c *fakeGoldenControl) goldenActorStatus() *ateapipb.ActorStatus { - st := &ateapipb.ActorStatus{State: c.goldenState} - if c.goldenSnapshot != "" { + st := &ateapipb.ActorStatus{State: c.goldenState, LastAssignedGeneration: c.generation} + if c.goldenState == ateapipb.ActorState_ACTOR_STATE_SUSPENDING && st.LastAssignedGeneration == 0 { st.LastAssignedGeneration = 1 + } + if c.goldenSnapshot != "" { + if st.LastAssignedGeneration == 0 { + st.LastAssignedGeneration = 1 + } st.Snapshots = []*ateapipb.Snapshot{ - newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, testTemplateUID, "golden-snapshot", c.goldenSnapshot, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED), + newDurableSnapshot(st.LastAssignedGeneration, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, testTemplateUID, "golden-snapshot", c.goldenSnapshot, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED), } + + } + if c.pausedLocal { + // The pause leaves the actor SUSPENDED with a node-local snapshot + // and no durable copy. + snap := newDurableSnapshot(st.LastAssignedGeneration, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, testTemplateUID, "pause-snapshot", "", ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) + snap.DurableSnapshot = nil + snap.Locality = "node-1" + st.Snapshots = append(st.Snapshots, snap) } return st } @@ -199,6 +241,9 @@ func (c *fakeGoldenControl) CreateActor(_ context.Context, req *ateapipb.CreateA // AlreadyExists means the actor does exist (e.g. a racing creation). if c.createErr == nil || apierror.Code(c.createErr) == codes.AlreadyExists { c.exists = true + c.actorTemplate = proto.CloneOf(req.GetActor().GetActorTemplate()) + c.generation = 0 + c.goldenSnapshot = "" c.goldenState = ateapipb.ActorState_ACTOR_STATE_SUSPENDED } if c.createErr != nil { @@ -207,8 +252,9 @@ func (c *fakeGoldenControl) CreateActor(_ context.Context, req *ateapipb.CreateA // Like the real Control service, return the stored actor: metadata plus // a status observing the initial SUSPENDED state. return &ateapipb.Actor{ - Metadata: req.GetActor().GetMetadata(), - Status: c.goldenActorStatus(), + Metadata: &ateapipb.ResourceMetadata{Atespace: req.GetActor().GetMetadata().GetAtespace(), Name: req.GetActor().GetMetadata().GetName(), Uid: c.goldenActorUID(), Version: c.goldenActorVersion()}, + Status: c.goldenActorStatus(), + ActorTemplate: c.goldenActorTemplate(), }, nil } @@ -222,8 +268,9 @@ func (c *fakeGoldenControl) GetActor(_ context.Context, req *ateapipb.GetActorRe return nil, apierror.NotFound("no such actor") } return &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: req.GetActor().GetAtespace(), Name: req.GetActor().GetName()}, - Status: c.goldenActorStatus(), + Metadata: &ateapipb.ResourceMetadata{Atespace: req.GetActor().GetAtespace(), Name: req.GetActor().GetName(), Uid: c.goldenActorUID(), Version: c.goldenActorVersion()}, + Status: c.goldenActorStatus(), + ActorTemplate: c.goldenActorTemplate(), }, nil } @@ -234,24 +281,15 @@ func (c *fakeGoldenControl) ResumeActor(_ context.Context, req *ateapipb.ResumeA if c.resumeErr != nil { return nil, c.resumeErr } - c.goldenState = ateapipb.ActorState_ACTOR_STATE_RUNNING + if c.goldenState != ateapipb.ActorState_ACTOR_STATE_RUNNING { + c.version = c.goldenActorVersion() + 1 + c.goldenState = ateapipb.ActorState_ACTOR_STATE_RUNNING + } return &ateapipb.ResumeActorResponse{}, nil } -func (c *fakeGoldenControl) SuspendActor(_ context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) { - c.mu.Lock() - defer c.mu.Unlock() - c.suspendReqs = append(c.suspendReqs, req) - if c.suspendErr != nil { - return nil, c.suspendErr - } - c.goldenState = ateapipb.ActorState_ACTOR_STATE_SUSPENDED - if c.goldenSnapshot == "" { - c.goldenSnapshot = c.snapshot - } - return &ateapipb.SuspendActorResponse{ - Actor: &ateapipb.Actor{Status: c.goldenActorStatus()}, - }, nil +func (c *fakeGoldenControl) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) { + return c.suspendActor(ctx, req, nil) } func (c *fakeGoldenControl) GetTag(_ context.Context, _ *ateapipb.GetTagRequest) (*ateapipb.Tag, error) { @@ -263,9 +301,12 @@ func (c *fakeGoldenControl) GetTag(_ context.Context, _ *ateapipb.GetTagRequest) return proto.CloneOf(c.tag), nil } -func (c *fakeGoldenControl) CreateTag(_ context.Context, req *ateapipb.CreateTagRequest) (*ateapipb.Tag, error) { +func (c *fakeGoldenControl) createTag(_ context.Context, req *ateapipb.CreateTagRequest, precondition *store.Precondition) (*ateapipb.Tag, error) { c.mu.Lock() defer c.mu.Unlock() + if precondition != nil && (precondition.UID != c.goldenActorUID() || precondition.Version != c.goldenActorVersion()) { + return nil, apierror.Aborted("actor revision changed") + } c.tagReqs = append(c.tagReqs, req) if c.tagErr != nil { return nil, c.tagErr @@ -303,6 +344,53 @@ func (c *fakeGoldenControl) DeleteActor(_ context.Context, req *ateapipb.DeleteA return &ateapipb.Actor{}, nil } +func (c *fakeGoldenControl) suspendActor(_ context.Context, req *ateapipb.SuspendActorRequest, precondition *store.Precondition) (*ateapipb.SuspendActorResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + if precondition != nil && (precondition.UID != c.goldenActorUID() || precondition.Version != c.goldenActorVersion()) { + return nil, apierror.Aborted("actor revision changed") + } + c.suspendReqs = append(c.suspendReqs, req) + if c.suspendErr != nil { + return nil, c.suspendErr + } + if c.goldenState == ateapipb.ActorState_ACTOR_STATE_RUNNING { + c.generation = c.goldenActorStatus().GetLastAssignedGeneration() + 1 + c.version = c.goldenActorVersion() + 1 + } else { + c.generation = c.goldenActorStatus().GetLastAssignedGeneration() + } + c.goldenState = ateapipb.ActorState_ACTOR_STATE_SUSPENDED + if c.goldenSnapshot == "" { + c.goldenSnapshot = c.snapshot + } + return &ateapipb.SuspendActorResponse{ + Actor: &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Atespace: req.GetActor().GetAtespace(), Name: req.GetActor().GetName(), Uid: c.goldenActorUID(), Version: c.goldenActorVersion()}, Status: c.goldenActorStatus(), ActorTemplate: c.goldenActorTemplate()}, + }, nil +} + +func (c *fakeGoldenControl) PauseActor(_ context.Context, req *ateapipb.PauseActorRequest) (*ateapipb.PauseActorResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.pauseReqs = append(c.pauseReqs, req) + if c.pauseErr != nil { + return nil, c.pauseErr + } + c.generation = c.goldenActorStatus().GetLastAssignedGeneration() + 1 + c.version = c.goldenActorVersion() + 1 + c.pausedLocal = true + c.goldenState = ateapipb.ActorState_ACTOR_STATE_SUSPENDED + return &ateapipb.PauseActorResponse{}, nil +} + +func (c *fakeGoldenControl) RevertActor(_ context.Context, _ *ateapipb.RevertActorRequest) (*ateapipb.RevertActorResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.goldenState = ateapipb.ActorState_ACTOR_STATE_SUSPENDED + c.version = c.goldenActorVersion() + 1 + return &ateapipb.RevertActorResponse{}, nil +} + func (c *fakeGoldenControl) callCounts() (creates, resumes, suspends int) { c.mu.Lock() defer c.mu.Unlock() @@ -357,7 +445,10 @@ func seededGoldenStatus(tmpl *ateapipb.ActorTemplate) *ateapipb.GoldenSnapshotSt func withSnapshotDeadline(at time.Time) func(*ateapipb.ActorTemplate) { return func(tmpl *ateapipb.ActorTemplate) { - seededGoldenStatus(tmpl).SnapshotTime = timestamppb.New(at) + status := seededGoldenStatus(tmpl) + status.SnapshotTime = timestamppb.New(at) + status.WarmupActorUid = someActorUID + status.WarmupActorVersion = 1 } } @@ -483,17 +574,20 @@ func TestReconcileOne(t *testing.T) { wantSuspends: 1, }, { - name: "suspending golden actor is completed and recorded", + name: "suspending golden actor is completed and rewarmed", template: testTemplate(), control: &fakeGoldenControl{exists: true, goldenState: ateapipb.ActorState_ACTOR_STATE_SUSPENDING, snapshot: goldenSnapshot}, wantTag: true, - wantSuspends: 1, + wantResumes: 1, + wantSuspends: 2, }, { - name: "suspended golden actor with a snapshot is recorded without more control calls", - template: testTemplate(), - control: &fakeGoldenControl{exists: true, goldenState: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, goldenSnapshot: goldenSnapshot}, - wantTag: true, + name: "suspended golden actor with a snapshot is rewarmed before capture", + template: testTemplate(), + control: &fakeGoldenControl{exists: true, goldenState: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, goldenSnapshot: goldenSnapshot}, + wantTag: true, + wantResumes: 1, + wantSuspends: 1, }, { name: "create AlreadyExists requeues for the retry to observe", @@ -956,8 +1050,8 @@ func TestReconcileOne_GoldenTagRecovery(t *testing.T) { if tagDurableSnapshotURI(control.tag, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED) == "" { t.Fatal("golden tag has no snapshot") } - if len(control.createReqs) != 0 || len(control.resumeReqs) != 0 || len(control.suspendReqs) != 0 { - t.Fatal("repeated golden actor warmup") + if tagDurableSnapshotURI(tt.tag, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED) != "" && (len(control.createReqs) != 0 || len(control.resumeReqs) != 0 || len(control.suspendReqs) != 0) { + t.Fatal("repeated warmup despite a completed tag") } if got := len(control.tagReqs); got != tt.wantCreateTags { t.Fatalf("CreateTag calls = %d, want %d", got, tt.wantCreateTags) diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index 9f86bc31d6..a18d260707 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -489,7 +489,7 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) { } return status.Error(codes.Unavailable, "worker pod is gone") } - if _, err := actorWorkflow.SuspendActor(ctx, actorRef, suspendSync); apierror.Code(err) != codes.Unavailable { + if _, err := actorWorkflow.SuspendActor(ctx, actorRef, suspendSync, nil); apierror.Code(err) != codes.Unavailable { t.Fatalf("SuspendActor err = %v, want Unavailable from the failed checkpoint", err) } diff --git a/cmd/ateapi/internal/controlapi/workflow_pause_test.go b/cmd/ateapi/internal/controlapi/workflow_pause_test.go index e07dc0429b..3cfb5c602d 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_pause_test.go @@ -61,7 +61,7 @@ func TestPauseActor_Finalized_WorkerGone(t *testing.T) { }) // Intentionally NOT creating the worker in store, simulates worker already gone. - if _, err := w.SuspendActor(ctx, actorRef, suspendAsync); apierror.Code(err) != codes.FailedPrecondition { + if _, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil); apierror.Code(err) != codes.FailedPrecondition { t.Fatalf("Async SuspendActor error = %v, want FailedPrecondition for the actor it crashed", err) } if checkpoints, _ := atelet.calls(); checkpoints != 1 { @@ -129,7 +129,7 @@ func TestPauseActor_Finalized_AlreadyCrashed(t *testing.T) { s.WorkerAssignment = nil }) - if _, err := w.SuspendActor(ctx, actorRef, suspendAsync); apierror.Code(err) != codes.FailedPrecondition { + if _, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil); apierror.Code(err) != codes.FailedPrecondition { t.Fatalf("Async SuspendActor error = %v, want FailedPrecondition for a CRASHED actor", err) } got := mustReadActor(t, ctx, st, actorRef) @@ -172,7 +172,7 @@ func TestPauseActor_Finalized_KeepsCrashFromWorkerDelete(t *testing.T) { seedWireWorker(t, ctx, st, assignmentOf(created)) w, _ := newWireCaptureWorkflow(t, workerDeletedDuringReleaseStore{Interface: st}) - _, err := w.SuspendActor(ctx, actorRef, suspendAsync) + _, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil) assertConcurrentCrashKept(t, ctx, st, actorRef, err, "suspend failed: "+crashMessageWorkerPodGone) } @@ -222,7 +222,7 @@ func TestPauseActor_Finalized_RecordsFidelity(t *testing.T) { }) seedWireWorker(t, ctx, st, assignmentOf(created)) - got, err := w.SuspendActor(ctx, actorRef, suspendAsync) + got, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil) if err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -289,7 +289,7 @@ func TestPauseActor_Finalized_MissingLocalSnapshotClearsAssignment(t *testing.T) s.Snapshots = nil }) - if _, err := w.SuspendActor(ctx, actorRef, suspendAsync); apierror.Code(err) != codes.FailedPrecondition { + if _, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil); apierror.Code(err) != codes.FailedPrecondition { t.Fatalf("Async SuspendActor error = %v, want FailedPrecondition for the actor it crashed", err) } got := mustReadActor(t, ctx, st, actorRef) @@ -357,7 +357,7 @@ func TestPauseActorWorkflow_RejectedAndIdempotentPaths(t *testing.T) { seedWorkflowActor(t, ctx, st, resources.ActorRef{Atespace: "team-a", Name: "id1"}, "ns", "tmpl1", tc.seedState, tc.seedOpts...) - actor, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendAsync) + actor, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendAsync, nil) if tc.wantErr { if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code(err) = %v, want %v (err: %v)", got, codes.FailedPrecondition, err) @@ -401,7 +401,7 @@ func TestPauseActor_MarkedPausing_StateMatrix(t *testing.T) { a.Status.WorkerAssignment = wireTestAssignment() }) - _, err := w.SuspendActor(ctx, actorRef, suspendAsync) + _, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil) stored := mustReadActor(t, ctx, st, actorRef) if seedState == ateapipb.ActorState_ACTOR_STATE_RUNNING { assertStoppedAtDial(t, err) @@ -442,7 +442,7 @@ func TestPauseActor_MarkedPausing_ReentrySkipsPauseInProgress(t *testing.T) { a.Status.AssignedNode = "node-1" a.Status.WorkerAssignment = wireTestAssignment() }) - paused, err := w.SuspendActor(ctx, actorRef, suspendAsync) + paused, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil) if err != nil { t.Fatalf("Async SuspendActor on a pause in progress: %v", err) } @@ -468,7 +468,7 @@ func TestPauseActor_MarkedPausing_ReentrySkipsPauseInProgress(t *testing.T) { newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "snap", "gs://snapshots/s", ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING), } }) - _, err = w.SuspendActor(ctx, suspendingRef, suspendAsync) + _, err = w.SuspendActor(ctx, suspendingRef, suspendAsync, nil) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Errorf("Async SuspendActor on a suspend in progress: code = %v, want FailedPrecondition (err: %v)", got, err) } @@ -522,7 +522,7 @@ func TestPauseActor_AteletPaused_DialFailureLeavesActorRetryable(t *testing.T) { storetest.MustCreateActor(t, ctx, persistence, actor) w := newMarkOnlyWorkflow(persistence) - _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, suspendAsync) + _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, suspendAsync, nil) assertStoppedAtDial(t, err) stored := mustReadActor(t, ctx, persistence, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}) @@ -552,7 +552,7 @@ func TestPauseActor_CrashesWhenPausingActorMissingWorkerPod(t *testing.T) { seedWorkflowActor(t, ctx, st, resources.ActorRef{Atespace: "team-a", Name: "id1"}, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDING, withPauseInProgress()) - _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendAsync) + _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendAsync, nil) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code(err) = %v, want %v (err: %v)", got, codes.FailedPrecondition, err) } @@ -582,7 +582,7 @@ func TestPauseActor_MarkedPausing_GoldenAtespaceRejected(t *testing.T) { a.Status.AssignedNode = "node-1" a.Status.WorkerAssignment = wireTestAssignment() }) - _, err := w.SuspendActor(ctx, actorRef, suspendAsync) + _, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code = %v (err %v), want FailedPrecondition", got, err) } @@ -628,7 +628,7 @@ func TestPauseActor_AteletPaused_ForwardsPreferredFidelity(t *testing.T) { }, }) - if _, err := w.SuspendActor(ctx, actorRef, suspendAsync); err != nil { + if _, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil); err != nil { t.Fatalf("SuspendActor: %v", err) } req := atelet.checkpointRequest() diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go index c745f7ba4f..12a09c0607 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go @@ -197,7 +197,7 @@ func TestRevertActor_ReturnsActorToItsSnapshot(t *testing.T) { }) seedWireWorker(t, ctx, st, assignmentOf(running)) wire, atelet := newWireCaptureWorkflow(t, st) - suspended, err := wire.SuspendActor(ctx, actorRef, suspendSync) + suspended, err := wire.SuspendActor(ctx, actorRef, suspendSync, nil) if err != nil { t.Fatalf("SuspendActor after revert: %v", err) } diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go index 665060350e..dee131784b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend.go @@ -72,8 +72,9 @@ func (m suspendMode) uploadMode() ateletpb.UploadMode { // takes a running actor: it is checkpointed on its worker and the snapshot is // kept on that node. Idempotent: a re-entered workflow fast-forwards past the // steps a previous attempt completed, deriving progress from the persisted -// actor alone. -func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.ActorRef, mode suspendMode) (_ *ateapipb.Actor, err error) { +// actor alone. A non-nil precondition pins the actor revision the suspend +// may act on, checked under the actor lease before any step runs. +func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.ActorRef, mode suspendMode, precondition *store.Precondition) (_ *ateapipb.Actor, err error) { ctx = preview.ContextWithOptIn(ctx) start := time.Now() var actor *ateapipb.Actor @@ -100,7 +101,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.Act } defer lease.Close() - actor, actorTemplate, err = w.loadActorForSuspend(leaseCtx, actorRef) + actor, actorTemplate, err = w.loadActorForSuspend(leaseCtx, actorRef, precondition) if err != nil { return nil, err } @@ -154,7 +155,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.Act } // loadActorForSuspend fetches the current actor record and its template. -func (w *ActorWorkflow) loadActorForSuspend(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, _ *ateapipb.ActorTemplate, err error) { +func (w *ActorWorkflow) loadActorForSuspend(ctx context.Context, actorRef resources.ActorRef, precondition *store.Precondition) (_ *ateapipb.Actor, _ *ateapipb.ActorTemplate, err error) { ctx, done := stepSpan(ctx, "LoadActorForSuspend") defer func() { err = done(err) }() @@ -162,6 +163,11 @@ func (w *ActorWorkflow) loadActorForSuspend(ctx context.Context, actorRef resour if err != nil { return nil, nil, err } + if precondition != nil { + if err := precondition.Check(actor.GetMetadata()); err != nil { + return nil, nil, apierror.Aborted("actor changed before suspend: %w", err) + } + } actorTemplate, err := resolveActorTemplate(ctx, w.store, actor) if err != nil { return nil, nil, err diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go b/cmd/ateapi/internal/controlapi/workflow_suspend_test.go index 80a3286242..aa04722e56 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend_test.go @@ -61,7 +61,7 @@ func TestSuspendActor_MarkedSuspending_SnapshotURI(t *testing.T) { }, }) w := newMarkOnlyWorkflow(persistence) - _, err := w.SuspendActor(ctx, actorRef, suspendSync) + _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) assertStoppedAtDial(t, err) marked := mustReadActor(t, ctx, persistence, actorRef) @@ -107,7 +107,7 @@ func TestSuspendActor_MarkedSuspending_ReentryKeepsPersistedSnapshotLocation(t * }, }, }) - suspended, err := w.SuspendActor(ctx, actorRef, suspendSync) + suspended, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -152,7 +152,7 @@ func TestSuspendActorWorkflow_RejectedAndIdempotentPaths(t *testing.T) { seedWorkflowActor(t, ctx, st, resources.ActorRef{Atespace: "team-a", Name: "id1"}, "ns", "tmpl1", tc.seedState) - actor, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendSync) + actor, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendSync, nil) if tc.wantErr { if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code(err) = %v, want %v (err: %v)", got, codes.FailedPrecondition, err) @@ -197,7 +197,7 @@ func TestSuspendActor_MarkedSuspending_StateMatrix(t *testing.T) { a.Status.WorkerAssignment = wireTestAssignment() }) - _, err := w.SuspendActor(ctx, actorRef, suspendSync) + _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) stored := mustReadActor(t, ctx, persistence, actorRef) switch seedState { case ateapipb.ActorState_ACTOR_STATE_RUNNING, ateapipb.ActorState_ACTOR_STATE_SUSPENDING: @@ -237,7 +237,7 @@ func TestSuspendActor_MarkedSuspending_RejectsPauseInProgress(t *testing.T) { a.Status.WorkerAssignment = wireTestAssignment() }) - _, err := w.SuspendActor(ctx, actorRef, suspendSync) + _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Errorf("SuspendActor on a pause in progress: code = %v, want FailedPrecondition (err: %v)", got, err) } @@ -276,7 +276,7 @@ func TestSuspendActor_MarkedSuspending_PausedPromotesLocalSnapshot(t *testing.T) }) // The upload's dial fails, so the actor stays as the mark left it. - _, err := w.SuspendActor(ctx, actorRef, suspendSync) + _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) assertStoppedAtDial(t, err) marked := mustReadActor(t, ctx, persistence, actorRef) if got := marked.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDING { @@ -321,7 +321,7 @@ func TestSuspendActor_CrashesWhenSuspendingActorMissingWorkerPod(t *testing.T) { seedWorkflowActor(t, ctx, st, resources.ActorRef{Atespace: "team-a", Name: "id1"}, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDING) - _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendSync) + _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, suspendSync, nil) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code(err) = %v, want %v (err: %v)", got, codes.FailedPrecondition, err) } @@ -400,7 +400,7 @@ func TestSuspendActor_AteletSuspended_DialFailureLeavesActorRetryable(t *testing storetest.MustCreateActor(t, ctx, persistence, actor) w := newMarkOnlyWorkflow(persistence) - _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, suspendSync) + _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "actor-1"}, suspendSync, nil) assertStoppedAtDial(t, err) // A dial failure is transient from this workflow's point of view: it @@ -452,7 +452,7 @@ func TestSuspendActor_AteletSuspended_WireRequest(t *testing.T) { }, }) - if _, err := w.SuspendActor(ctx, actorRef, suspendSync); err != nil { + if _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil); err != nil { t.Fatalf("SuspendActor: %v", err) } req := atelet.checkpointRequest() @@ -505,7 +505,7 @@ func TestSuspendActor_Finalized_NoAssignment(t *testing.T) { }, }) - stored, err := w.SuspendActor(ctx, actorRef, suspendSync) + stored, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -613,7 +613,7 @@ func TestSuspendActor_Finalized_ReleasesReplacedSnapshot(t *testing.T) { } }) - stored, err := w.SuspendActor(ctx, actorRef, suspendSync) + stored, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -692,7 +692,7 @@ func TestSuspendActor_Finalized_PauseReleasesReplacedSnapshot(t *testing.T) { Object: &ateletpb.ObjectStorage{SnapshotUri: paused.String()}, }}} - stored, err := w.SuspendActor(ctx, actorRef, suspendAsync) + stored, err := w.SuspendActor(ctx, actorRef, suspendAsync, nil) if err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -746,7 +746,7 @@ func TestSuspendActor_PausedSnapshotAlreadyUploaded(t *testing.T) { } }) - stored, err := w.SuspendActor(ctx, actorRef, suspendSync) + stored, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -815,7 +815,7 @@ func TestSuspendActor_Finalized_CommitsDespiteObjectStoreFailure(t *testing.T) { }) objects.OnDelete = func(string, string) error { return errObjectStore } - stored, err := w.SuspendActor(ctx, actorRef, suspendSync) + stored, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -872,7 +872,7 @@ func TestSuspendActor_Finalized_KeepsReplacedSnapshotOnConflict(t *testing.T) { } }) - if _, err := w.SuspendActor(ctx, actorRef, suspendSync); apierror.Code(err) != codes.Aborted { + if _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil); apierror.Code(err) != codes.Aborted { t.Fatalf("SuspendActor = %v, want code Aborted", err) } // Every actor update conflicts, so pin that the Aborted came from finalize @@ -921,7 +921,7 @@ func TestSuspendActor_Finalized_KeepsCrashFromWorkerDelete(t *testing.T) { seedWireWorker(t, ctx, persistence, assignmentOf(created)) w, _ := newWireCaptureWorkflow(t, workerDeletedDuringReleaseStore{Interface: persistence}) - _, err := w.SuspendActor(ctx, actorRef, suspendSync) + _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) assertConcurrentCrashKept(t, ctx, persistence, actorRef, err, "suspend failed: "+crashMessageWorkerPodGone) } @@ -980,7 +980,7 @@ func TestSuspendActor_Finalized_ReleasesOnlyOwnWorker(t *testing.T) { ActorUid: uid, }) - if _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "shared"}, suspendSync); err != nil { + if _, err := w.SuspendActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "shared"}, suspendSync, nil); err != nil { t.Fatalf("SuspendActor: %v", err) } @@ -1085,7 +1085,7 @@ func TestSuspendActor_PausedSnapshotUploaded_Preconditions(t *testing.T) { }, }) - _, err := w.SuspendActor(ctx, actorRef, suspendSync) + _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code(err) = %v, want %v for a missing node record (err: %v)", got, codes.FailedPrecondition, err) } @@ -1119,7 +1119,7 @@ func TestSuspendActor_PausedSnapshotUploaded_Preconditions(t *testing.T) { }, }) - _, err := w.SuspendActor(ctx, actorRef, suspendSync) + _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil) if !errors.Is(err, ErrNoAteletOnNode) { t.Fatalf("SuspendActor = %v, want ErrNoAteletOnNode", err) } @@ -1157,7 +1157,7 @@ func TestSuspendActor_PausedSnapshotUploaded_WireRequest(t *testing.T) { }, }) - if _, err := w.SuspendActor(ctx, actorRef, suspendSync); err != nil { + if _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil); err != nil { t.Fatalf("SuspendActor: %v", err) } req := atelet.promoteRequest() @@ -1205,7 +1205,7 @@ func TestSuspendActor_Finalized_MissingDurableSnapshotClearsAssignment(t *testin s.Snapshots = nil }) - if _, err := w.SuspendActor(ctx, actorRef, suspendSync); apierror.Code(err) != codes.FailedPrecondition { + if _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil); apierror.Code(err) != codes.FailedPrecondition { t.Fatalf("SuspendActor error = %v, want FailedPrecondition for the actor it crashed", err) } got, err := st.GetActor(ctx, actorRef) @@ -1452,7 +1452,7 @@ func TestSuspendAndPause_CharacteristicMatrix(t *testing.T) { seedWireWorker(t, ctx, persistence, assignment) w, atelet := newWireCaptureWorkflow(t, persistence) - _, err := w.SuspendActor(ctx, actorRef, mode.suspendMode) + _, err := w.SuspendActor(ctx, actorRef, mode.suspendMode, nil) stored := mustReadActor(t, ctx, persistence, actorRef) got := suspendOutcome{ code: apierror.Code(err), @@ -1506,7 +1506,7 @@ func TestSuspendAndPause_FinalizeWithoutAssignment(t *testing.T) { s.WorkerAssignment = nil }) - returned, err := w.SuspendActor(ctx, actorRef, mode.suspendMode) + returned, err := w.SuspendActor(ctx, actorRef, mode.suspendMode, nil) if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("%s: code = %v (err %v), want FailedPrecondition", mode.name, got, err) } @@ -1545,7 +1545,7 @@ func TestSuspendAndPause_InvalidStorageLocationRejected(t *testing.T) { }, }) - if _, err := w.SuspendActor(ctx, actorRef, mode.suspendMode); err == nil { + if _, err := w.SuspendActor(ctx, actorRef, mode.suspendMode, nil); err == nil { t.Fatalf("%s: expected error for invalid storage location, got nil", mode.name) } stored := mustReadActor(t, ctx, persistence, actorRef) @@ -1770,7 +1770,7 @@ func TestSuspendActor_AppliesCheckpointState(t *testing.T) { return nil } - _, err := w.SuspendActor(ctx, actorRef, tt.mode) + _, err := w.SuspendActor(ctx, actorRef, tt.mode, nil) stored := mustReadActor(t, ctx, persistence, actorRef) if tt.wantCode != codes.OK { if got := apierror.Code(err); got != tt.wantCode { @@ -1830,7 +1830,7 @@ func TestSuspendActor_AppliesPromoteSnapshotState(t *testing.T) { Object: &ateletpb.ObjectStorage{SnapshotUri: uri.String()}, }}} - if _, err := w.SuspendActor(ctx, actorRef, suspendSync); err != nil { + if _, err := w.SuspendActor(ctx, actorRef, suspendSync, nil); err != nil { t.Fatalf("SuspendActor: %v", err) } if checkpoints, promotes := atelet.calls(); checkpoints != 0 || promotes != 1 { @@ -1841,3 +1841,35 @@ func TestSuspendActor_AppliesPromoteSnapshotState(t *testing.T) { t.Errorf("snapshots mismatch (-want +got):\n%s", diff) } } + +func TestSuspendActor_RevisionPreconditions(t *testing.T) { + for _, name := range []string{"stale uid", "stale version", "current revision"} { + t.Run(name, func(t *testing.T) { + ctx := t.Context() + st, cleanup := storetest.SetupTestStore(t) + defer cleanup() + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + ref := resources.ActorRef{Atespace: "team-a", Name: "id1"} + seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDED) + actor, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatal(err) + } + precondition := store.PreconditionFrom(actor) + if name == "stale uid" { + precondition.UID = "22222222-2222-4222-8222-222222222222" + } + if name == "stale version" { + precondition.Version++ + } + _, err = w.SuspendActor(ctx, ref, suspendSync, &precondition) + if name == "current revision" { + if err != nil { + t.Fatal(err) + } + } else if apierror.Code(err) != codes.Aborted { + t.Fatalf("suspend with %s = %v, want Aborted", name, err) + } + }) + } +} diff --git a/cmd/ateapi/internal/controlapi/workflow_tag.go b/cmd/ateapi/internal/controlapi/workflow_tag.go index 690d52bc84..17d4402769 100644 --- a/cmd/ateapi/internal/controlapi/workflow_tag.go +++ b/cmd/ateapi/internal/controlapi/workflow_tag.go @@ -41,13 +41,13 @@ import ( // // The tag captures whichever snapshot the actor holds when the workflow runs. // An actor keeps no snapshot history, so a suspend that lands first moves what -// gets tagged; that race is inherent to naming an actor rather than a snapshot. +// gets tagged. A caller can pin the observed Actor revision with a precondition. // // Not idempotent: the name is taken as soon as phase 1 lands, so a create that // dies after it leaves a pending tag and every later create under that name is // AlreadyExists. To retry, delete the tag, which collects whatever the failed // attempt stranded, and create it again. -func (w *ActorWorkflow) TagActorSnapshot(ctx context.Context, tag *ateapipb.Tag) (*ateapipb.Tag, error) { +func (w *ActorWorkflow) TagActorSnapshot(ctx context.Context, tag *ateapipb.Tag, precondition *store.Precondition) (*ateapipb.Tag, error) { ctx = preview.ContextWithOptIn(ctx) actorRef := resources.ActorRefFromObjectRef(tag.GetSourceActor()) @@ -68,7 +68,7 @@ func (w *ActorWorkflow) TagActorSnapshot(ctx context.Context, tag *ateapipb.Tag) } defer tagLease.Close() - actor, actorTemplate, err := w.loadActorForTag(leaseCtx, actorRef) + actor, actorTemplate, err := w.loadActorForTag(leaseCtx, actorRef, precondition) if err != nil { return nil, err } @@ -205,7 +205,7 @@ func (w *ActorWorkflow) finalizeTagDeleted(ctx context.Context, tagRef resources // loadActorForTag fetches the actor to tag and its template, and checks that // the actor holds an external snapshot a tag can be made from. -func (w *ActorWorkflow) loadActorForTag(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, _ *ateapipb.ActorTemplate, err error) { +func (w *ActorWorkflow) loadActorForTag(ctx context.Context, actorRef resources.ActorRef, precondition *store.Precondition) (_ *ateapipb.Actor, _ *ateapipb.ActorTemplate, err error) { ctx, done := stepSpan(ctx, "LoadActorForTag") defer func() { err = done(err) }() @@ -213,6 +213,11 @@ func (w *ActorWorkflow) loadActorForTag(ctx context.Context, actorRef resources. if err != nil { return nil, nil, err } + if precondition != nil { + if err := precondition.Check(actor.GetMetadata()); err != nil { + return nil, nil, apierror.Aborted("source actor changed before tagging: %w", err) + } + } // Only a suspended actor's snapshot is complete. A running or // suspending actor's is either stale or still being written. if got := actor.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { diff --git a/cmd/ateapi/internal/controlapi/workflow_tag_test.go b/cmd/ateapi/internal/controlapi/workflow_tag_test.go index 425bd24bd4..ce36bcb283 100644 --- a/cmd/ateapi/internal/controlapi/workflow_tag_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_tag_test.go @@ -85,7 +85,7 @@ func TestTagActorSnapshot(t *testing.T) { actor, actorSnapshot := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst") actorRef := resources.ActorRefFromActor(actor) - tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if err != nil { t.Fatalf("TagActorSnapshot: %v", err) } @@ -140,7 +140,7 @@ func TestTagActorSnapshot_ActorRepointedToAnotherTemplate(t *testing.T) { t.Fatalf("repointing actor %s: %v", actorRef, err) } - tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if err != nil { t.Fatalf("TagActorSnapshot: %v", err) } @@ -230,7 +230,7 @@ func TestTagActorSnapshot_Preconditions(t *testing.T) { }) } - _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if code := apierror.Code(err); code != tt.wantCode { t.Fatalf("TagActorSnapshot error = %v (code %v), want code %v", err, code, tt.wantCode) } @@ -291,7 +291,7 @@ func TestTagActorSnapshot_TagsNewestExternalSnapshot(t *testing.T) { s.Snapshots = []*ateapipb.Snapshot{snap} }) - if _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")); err != nil { + if _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil); err != nil { t.Fatalf("TagActorSnapshot: %v", err) } }) @@ -320,7 +320,7 @@ func TestTagActorSnapshot_RecreateAfterCopyFailure(t *testing.T) { } return nil } - if _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")); !isObjectStoreErr(err) { + if _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil); !isObjectStoreErr(err) { t.Fatalf("TagActorSnapshot = %v, want an error wrapping %v", err, errObjectStore) } objects.OnCopy = nil @@ -340,7 +340,7 @@ func TestTagActorSnapshot_RecreateAfterCopyFailure(t *testing.T) { // Creating under the same name again is refused, even for the actor the // pending row was reserved for, and leaves that row exactly as it was. - _, err = w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + _, err = w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if code := apierror.Code(err); code != codes.AlreadyExists { t.Fatalf("TagActorSnapshot over the pending tag = %v (code %v), want code AlreadyExists", err, code) } @@ -364,7 +364,7 @@ func TestTagActorSnapshot_RecreateAfterCopyFailure(t *testing.T) { } // The create now runs from scratch, into a prefix of its own. - tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if err != nil { t.Fatalf("TagActorSnapshot after the delete: %v", err) } @@ -380,7 +380,7 @@ func TestTagActorSnapshot_RecreateAfterCopyFailure(t *testing.T) { } // A create over the finished tag is refused the same way. - _, err = w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + _, err = w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if code := apierror.Code(err); code != codes.AlreadyExists { t.Errorf("TagActorSnapshot over the finished tag = %v (code %v), want code AlreadyExists", err, code) } @@ -416,7 +416,7 @@ func TestTagActorSnapshot_RecreateAfterRecordFailure(t *testing.T) { actorRef := resources.ActorRefFromActor(actor) tagRef := resources.TagRef{Atespace: "team-a", Name: "v1"} - if _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")); !errors.Is(err, failing.err) { + if _, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil); !errors.Is(err, failing.err) { t.Fatalf("TagActorSnapshot = %v, want an error wrapping %v", err, failing.err) } failing.err = nil @@ -445,7 +445,7 @@ func TestTagActorSnapshot_RecreateAfterRecordFailure(t *testing.T) { t.Errorf("GetTag after the delete = %v, want ErrNotFound", err) } - tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + tag, err := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if err != nil { t.Fatalf("TagActorSnapshot after the delete: %v", err) } @@ -484,7 +484,7 @@ func TestTagActorSnapshot_RacesDelete(t *testing.T) { return nil } - tag, createErr := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1")) + tag, createErr := w.TagActorSnapshot(ctx, tagToCreate(actorRef, "v1"), nil) if pendingURI == "" { t.Fatalf("the create never reserved a row to race with (TagActorSnapshot = %v)", createErr) } @@ -531,7 +531,7 @@ func TestTagActorSnapshot_NameTakenByAnotherActor(t *testing.T) { } return nil } - if _, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(first), "v1")); !isObjectStoreErr(err) { + if _, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(first), "v1"), nil); !isObjectStoreErr(err) { t.Fatalf("TagActorSnapshot(actor-1) = %v, want an error wrapping %v", err, errObjectStore) } pending, err := persistence.GetTag(ctx, tagRef) @@ -543,7 +543,7 @@ func TestTagActorSnapshot_NameTakenByAnotherActor(t *testing.T) { // The second actor asks for the same name, with object storage healthy: it // must not inherit the first actor's prefix. objects.OnCopy = nil - _, err = w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(second), "v1")) + _, err = w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(second), "v1"), nil) if code := apierror.Code(err); code != codes.AlreadyExists { t.Fatalf("TagActorSnapshot(actor-2) = %v (code %v), want code AlreadyExists", err, code) } @@ -572,7 +572,7 @@ func TestDeleteTag_ReleasesExternalSnapshot(t *testing.T) { template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") w, objects := newFinalizeWorkflow(persistence) actor, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst") - tag, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(actor), "v1")) + tag, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(actor), "v1"), nil) if err != nil { t.Fatalf("TagActorSnapshot: %v", err) } @@ -678,7 +678,7 @@ func TestDeleteTag_Preconditions(t *testing.T) { template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") w, objects := newFinalizeWorkflow(persistence) actor, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst") - tag, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(actor), "v1")) + tag, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(actor), "v1"), nil) if err != nil { t.Fatalf("TagActorSnapshot: %v", err) } @@ -715,3 +715,38 @@ func TestDeleteTag_Preconditions(t *testing.T) { t.Errorf("GetTag after the delete = %v, want ErrNotFound", err) } } + +func TestTagActorSnapshot_RevisionPreconditions(t *testing.T) { + for _, name := range []string{"stale uid", "stale version", "resumed actor", "current revision"} { + t.Run(name, func(t *testing.T) { + ctx := t.Context() + persistence := newTestPersistence(t) + template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") + w, objects := newFinalizeWorkflow(persistence) + actor, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json") + precondition := store.PreconditionFrom(actor) + if name == "stale uid" { + precondition.UID = "22222222-2222-4222-8222-222222222222" + } + if name == "stale version" { + precondition.Version++ + } + if name == "resumed actor" { + mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { s.State = ateapipb.ActorState_ACTOR_STATE_RUNNING }) + } + _, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(actor), "v1"), &precondition) + if name == "current revision" { + if err != nil { + t.Fatal(err) + } + } else { + if apierror.Code(err) != codes.Aborted { + t.Fatalf("tag with %s = %v, want Aborted", name, err) + } + if _, err := persistence.GetTag(ctx, resources.TagRef{Atespace: actor.GetMetadata().GetAtespace(), Name: "v1"}); !errors.Is(err, store.ErrNotFound) { + t.Fatal("stale tag request reserved a name:", err) + } + } + }) + } +} diff --git a/cmd/ateapi/internal/controlapi/workflow_test.go b/cmd/ateapi/internal/controlapi/workflow_test.go index 9edc39e3b0..320eb82ea4 100644 --- a/cmd/ateapi/internal/controlapi/workflow_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_test.go @@ -76,7 +76,7 @@ func TestActorStateChangeRecords(t *testing.T) { }, }, transition: func(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, _ *ateapipb.Actor, _ *ateapipb.ActorTemplate) { - if _, err := w.SuspendActor(context.Background(), ref, suspendAsync); err != nil { + if _, err := w.SuspendActor(context.Background(), ref, suspendAsync, nil); err != nil { t.Fatalf("SuspendActor: %v", err) } }, @@ -181,7 +181,7 @@ func markOnly(t *testing.T, w *ActorWorkflow, ref resources.ActorRef, actor *ate t.Fatalf("UpdateActor: %v", err) } w.dialer = newDanglingDialer() - _, err := w.SuspendActor(ctx, ref, mode) + _, err := w.SuspendActor(ctx, ref, mode, nil) assertStoppedAtDial(t, err) } diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index 56f840facb..92b787eb73 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -2860,8 +2860,9 @@ type GoldenSnapshotStatus struct { // +k8s:optional // +k8s:subfield(atespace)=+k8s:required GoldenTag *ObjectRef `protobuf:"bytes,1,opt,name=golden_tag,json=goldenTag,proto3" json:"golden_tag,omitempty"` - // snapshot_time is when the golden-actor warmup ends and the - // golden snapshot may be taken. + // snapshot_time is when uninterrupted warmup ends for the Actor identified + // by warmup_actor_uid and warmup_actor_version. It is set once RUNNING is + // observed and stays unset otherwise; capture waits for it. // // +k8s:optional SnapshotTime *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=snapshot_time,json=snapshotTime,proto3" json:"snapshot_time,omitempty"` @@ -2870,9 +2871,18 @@ type GoldenSnapshotStatus struct { // // +k8s:optional // +k8s:maxLength=4096 - ErrorMessage string `protobuf:"bytes,3,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + ErrorMessage string `protobuf:"bytes,3,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` + // warmup_actor_uid identifies the Actor whose RUNNING state starts the timer. + // +k8s:optional + // +k8s:format=k8s-uuid + WarmupActorUid string `protobuf:"bytes,4,opt,name=warmup_actor_uid,json=warmupActorUid,proto3" json:"warmup_actor_uid,omitempty"` + // warmup_actor_version is the RUNNING revision whose warmup is being timed. + // A different revision starts a fresh timer, even if the Actor is RUNNING again. + // +k8s:optional + // +k8s:minimum=1 + WarmupActorVersion int64 `protobuf:"varint,5,opt,name=warmup_actor_version,json=warmupActorVersion,proto3" json:"warmup_actor_version,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *GoldenSnapshotStatus) Reset() { @@ -2926,6 +2936,20 @@ func (x *GoldenSnapshotStatus) GetErrorMessage() string { return "" } +func (x *GoldenSnapshotStatus) GetWarmupActorUid() string { + if x != nil { + return x.WarmupActorUid + } + return "" +} + +func (x *GoldenSnapshotStatus) GetWarmupActorVersion() int64 { + if x != nil { + return x.WarmupActorVersion + } + return 0 +} + type ActorTemplateStatus struct { state protoimpl.MessageState `protogen:"open.v1"` // +k8s:optional @@ -9580,12 +9604,14 @@ const file_ateapi_proto_rawDesc = "" + "\x06limits\x18\x01 \x03(\v2\x0e.ateapi.LimitsR\x06limits\"8\n" + "\x06Limits\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x1a\n" + - "\bquantity\x18\x02 \x01(\tR\bquantity\"\xae\x01\n" + + "\bquantity\x18\x02 \x01(\tR\bquantity\"\x8a\x02\n" + "\x14GoldenSnapshotStatus\x120\n" + "\n" + "golden_tag\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\tgoldenTag\x12?\n" + "\rsnapshot_time\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\fsnapshotTime\x12#\n" + - "\rerror_message\x18\x03 \x01(\tR\ferrorMessage\"i\n" + + "\rerror_message\x18\x03 \x01(\tR\ferrorMessage\x12(\n" + + "\x10warmup_actor_uid\x18\x04 \x01(\tR\x0ewarmupActorUid\x120\n" + + "\x14warmup_actor_version\x18\x05 \x01(\x03R\x12warmupActorVersion\"i\n" + "\x13ActorTemplateStatus\x12R\n" + "\x16golden_snapshot_status\x18\x01 \x01(\v2\x1c.ateapi.GoldenSnapshotStatusR\x14goldenSnapshotStatus\"\xd8\x01\n" + "\x0eSnapshotConfig\x12)\n" + diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index bbb5e5e3c0..b6edafbbb7 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -1200,8 +1200,9 @@ message GoldenSnapshotStatus { // +k8s:subfield(atespace)=+k8s:required ObjectRef golden_tag = 1; - // snapshot_time is when the golden-actor warmup ends and the - // golden snapshot may be taken. + // snapshot_time is when uninterrupted warmup ends for the Actor identified + // by warmup_actor_uid and warmup_actor_version. It is set once RUNNING is + // observed and stays unset otherwise; capture waits for it. // // +k8s:optional google.protobuf.Timestamp snapshot_time = 2; @@ -1212,6 +1213,17 @@ message GoldenSnapshotStatus { // +k8s:optional // +k8s:maxLength=4096 string error_message = 3; + + // warmup_actor_uid identifies the Actor whose RUNNING state starts the timer. + // +k8s:optional + // +k8s:format=k8s-uuid + string warmup_actor_uid = 4; + + // warmup_actor_version is the RUNNING revision whose warmup is being timed. + // A different revision starts a fresh timer, even if the Actor is RUNNING again. + // +k8s:optional + // +k8s:minimum=1 + int64 warmup_actor_version = 5; } message ActorTemplateStatus {