UpdateActor fails with Aborted while another operation on the actor is in progress

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()

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 {

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.