diff --git a/cmd/ateapi/internal/controlapi/actor.go b/cmd/ateapi/internal/controlapi/actor.go index ecdef8e9f9..ffee8080ea 100644 --- a/cmd/ateapi/internal/controlapi/actor.go +++ b/cmd/ateapi/internal/controlapi/actor.go @@ -257,6 +257,12 @@ func (s *RPCService) UpdateActor(ctx context.Context, req *ateapi.UpdateActorReq actorRef := resources.ActorRefFromActor(inActor) setSpanActorRefAttributes(ctx, actorRef) + ctx, lease, err := acquireLease(ctx, s.impl, actorLeaseKey(actorRef), "actor") + if err != nil { + return nil, err + } + defer lease.Close() + storedActor, err := s.impl.UpdateActor(ctx, actorRef, store.PreconditionFrom(inActor), func(toUpdate *ateapi.Actor) error { // Status and Metadata are server-owned fields. status, metadata := toUpdate.GetStatus(), toUpdate.GetMetadata() diff --git a/cmd/ateapi/internal/controlapi/actor_test.go b/cmd/ateapi/internal/controlapi/actor_test.go index d6296b1f10..29b4057422 100644 --- a/cmd/ateapi/internal/controlapi/actor_test.go +++ b/cmd/ateapi/internal/controlapi/actor_test.go @@ -139,6 +139,9 @@ func TestUpdateActor(t *testing.T) { req *ateapi.Actor want *ateapi.Actor wantCode codes.Code + // leaseConflict holds the actor lease across the update, + // as a running workflow does. + leaseConflict bool }{ { name: "sets a worker_selector the stored actor does not have", @@ -206,6 +209,16 @@ func TestUpdateActor(t *testing.T) { }, wantCode: codes.InvalidArgument, }, + { + name: "refused while another operation holds the actor", + stored: &ateapi.Actor{}, + req: &ateapi.Actor{ + ActorTemplate: &ateapi.ObjectRef{Atespace: templateNS, Name: templateName}, + WorkerSelector: &ateapi.Selector{MatchLabels: map[string]string{"tier": "paid"}}, + }, + leaseConflict: true, + wantCode: codes.Aborted, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -216,6 +229,15 @@ func TestUpdateActor(t *testing.T) { } svc, created := rpcServiceWithActor(t, tt.stored) + actorRef := resources.ActorRefFromActor(created) + + if tt.leaseConflict { + lease, err := svc.impl.AcquireLease(context.Background(), actorLeaseKey(actorRef)) + if err != nil { + t.Fatalf("AcquireLease() failed: %v", err) + } + defer lease.Close() + } tt.req.Metadata = created.GetMetadata() updated, err := svc.UpdateActor(context.Background(), &ateapi.UpdateActorRequest{Actor: tt.req}) @@ -224,6 +246,13 @@ func TestUpdateActor(t *testing.T) { if code := apierror.Code(err); code != tt.wantCode { t.Errorf("UpdateActor error = %v (code %v), want code %v", err, code, tt.wantCode) } + got, err := svc.impl.GetActor(context.Background(), actorRef) + if err != nil { + t.Fatal(err) + } + if got.GetMetadata().GetVersion() != created.GetMetadata().GetVersion() { + t.Errorf("a refused update wrote the actor: version %d, want %d", got.GetMetadata().GetVersion(), created.GetMetadata().GetVersion()) + } return } if err != nil { diff --git a/docs/api-guide.md b/docs/api-guide.md index 776bf826ce..5d98d077b5 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -657,10 +657,10 @@ Replaces the mutable fields of an existing actor with the ones in the request (` * **Request:** `UpdateActorRequest` * `actor`: `Actor` — the complete replacement actor. `metadata.atespace` and `metadata.name` identify the resource; `metadata.uid` and `metadata.version` are **required** preconditions. `metadata` and `status` are server-owned and whatever the request carries in them is ignored. `source_tag` is immutable. * `actor_template` may change only while the actor is `SUSPENDED`. The new template must declare the same volumes and volume mounts. Once the actor owns an external snapshot, the new template must also store snapshots under the same location: its `snapshotConfig.storageLocation` with the `PerTemplateSnapshotLocation` preview gate, otherwise ate-api-server's `--snapshot-location` (see [Snapshot Storage Layout](#snapshot-storage-layout)). - * `worker_selector` may be updated or cleared (`nil`) in any state; the updated selector takes effect on the next `ResumeActor`. + * `worker_selector` may be updated or cleared (`nil`) in any state, but not while another operation is changing the actor. The updated selector takes effect on the next `ResumeActor`. * A suspended actor with a local snapshot is initially pinned to its node (see `locality` in [Snapshot Storage Layout](#snapshot-storage-layout)). `ResumeActor` first looks for a worker there matching the template and actor selectors. If none is available but a worker satisfies the remaining constraints without node or runtime restrictions, resume detaches the checkpoint and retries placement. If no eligible worker has room, it returns `RESOURCE_EXHAUSTED`. * **Response:** the updated `Actor`. -* **Errors:** `INVALID_ARGUMENT` if `uid` or `version` is unset, or if the request changes an immutable field — including by leaving one unset; `FAILED_PRECONDITION` if a change to `actor_template` breaks one of the rules above; `ABORTED` if either guard no longer matches the stored resource. +* **Errors:** `INVALID_ARGUMENT` if `uid` or `version` is unset, or if the request changes an immutable field — including by leaving one unset; `FAILED_PRECONDITION` if a change to `actor_template` breaks one of the rules above; `ABORTED` if either guard no longer matches the stored resource, or while another operation is changing the actor (retry the same request). Because the guards are required and only a read supplies them, an update is always a read-modify-write. To Update an `Actor`, you must first `GetActor`/`CreateActor`, instead of building a new one — see [§7.2 of the API style guide](api-style-guide.md#72-using-version-and-uid-to-guard-writes) for why reconstructing the message can silently drop data.