From 9e20e39a877ca2674fafa78d6480f7f6bbc60345 Mon Sep 17 00:00:00 2001 From: Ivy Date: Wed, 14 Oct 2026 13:03:40 -0700 Subject: [PATCH 1/2] ateapi: apply atelet's snapshot state reports to an actor With asynchronous suspend, atelet uploads a suspended actor's snapshot and later detaches the actor from its node on its own timers. The control plane learns about both from atelet's reports. ApplyReportedSnapshotState applies the two facts a report carries: - An upload finished. If latest_durable names a recorded snapshot with the same uuid and upload URI, that snapshot becomes COMPLETED, with its manifest's fidelity. A generation at or below the latest COMPLETED one is stale and changes nothing. - The node detached the actor, shown by a report with neither latest_local nor uploading. It covers the node's snapshots up to the reported latest_durable generation. A newer snapshot postdates the detach, so a late or repeated report leaves it alone. Covered localities are cleared, and a suspended actor with no snapshot left on the node is released from it. A detach is refused while the latest covered snapshot isn't durable, so the actor can't resume from an older one. A report applies as a whole or not at all. Reports apply under the actor's lease, and only to a SUSPENDED or RUNNING actor that is the named incarnation and is on the reporting node. That check also runs before the lease is taken, so an atelet can't hold the lease of an actor on another node. A report that changes nothing writes nothing. --- .../internal/controlapi/snapshot_report.go | 204 ++++++++ .../controlapi/snapshot_report_test.go | 439 ++++++++++++++++++ 2 files changed, 643 insertions(+) create mode 100644 cmd/ateapi/internal/controlapi/snapshot_report.go create mode 100644 cmd/ateapi/internal/controlapi/snapshot_report_test.go diff --git a/cmd/ateapi/internal/controlapi/snapshot_report.go b/cmd/ateapi/internal/controlapi/snapshot_report.go new file mode 100644 index 0000000000..8efd8a3fa8 --- /dev/null +++ b/cmd/ateapi/internal/controlapi/snapshot_report.go @@ -0,0 +1,204 @@ +// 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" + "log/slog" + + "google.golang.org/protobuf/proto" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/apierror" + "github.com/agent-substrate/substrate/internal/preview" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +// ApplyActorSnapshotState records the snapshot state that the atelet on node +// reports for the Actor. The caller has authenticated that atelet. +func (s *RPCService) ApplyActorSnapshotState(ctx context.Context, actorRef resources.ActorRef, actorUID, node string, state *ateapipb.ActorSnapshotState) error { + setSpanActorRefAttributes(ctx, actorRef) + return s.actorWorkflow.ApplyReportedSnapshotState(ctx, actorRef, actorUID, node, state) +} + +// ApplyReportedSnapshotState applies a report from the atelet on node to the +// Actor. It holds the Actor's lease and applies reports only to a SUSPENDED or +// RUNNING Actor, so a report never interleaves with a workflow. A report that +// changes nothing writes nothing. +func (w *ActorWorkflow) ApplyReportedSnapshotState(ctx context.Context, actorRef resources.ActorRef, actorUID, node string, state *ateapipb.ActorSnapshotState) error { + ctx = preview.ContextWithOptIn(ctx) + // Check before taking the lease, so an atelet can't hold the lease of an + // Actor that isn't on its node. Check again under the lease. + if _, err := w.loadReportedActor(ctx, actorRef, actorUID, node); err != nil { + return err + } + leaseCtx, lease, err := w.acquireActorLease(ctx, actorRef) + if err != nil { + return err + } + defer lease.Close() + before, err := w.loadReportedActor(leaseCtx, actorRef, actorUID, node) + if err != nil { + return err + } + + // A workflow owns a transitioning Actor's snapshots, even after its lease + // lapses. A crashed Actor's snapshots wait for a revert. + switch st := before.GetStatus().GetState(); st { + case ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ateapipb.ActorState_ACTOR_STATE_RUNNING: + case ateapipb.ActorState_ACTOR_STATE_CRASHED: + return apierror.FailedPrecondition("actor %s is %s; revert it first", actorRef, st) + case ateapipb.ActorState_ACTOR_STATE_SUSPENDING, ateapipb.ActorState_ACTOR_STATE_RESUMING, + ateapipb.ActorState_ACTOR_STATE_DELETING, ateapipb.ActorState_ACTOR_STATE_REVERTING: + return apierror.Aborted("actor %s is %s; retry once that finishes", actorRef, st) + default: + return apierror.Internal("actor %s has unexpected state %s", actorRef, st) + } + + applied := proto.Clone(before).(*ateapipb.Actor) + if err := applyActorSnapshotState(applied.GetStatus(), node, state); err != nil { + return err + } + if proto.Equal(applied.GetStatus(), before.GetStatus()) { + return nil + } + if _, err := w.store.UpdateActor(leaseCtx, actorRef, store.PreconditionFrom(before), func(toUpdate *ateapipb.Actor) error { + toUpdate.Status = applied.GetStatus() + return nil + }); err != nil { + if errors.Is(err, store.ErrVersionConflict) { + return apierror.Aborted("concurrent update conflict, please retry") + } + return err + } + return nil +} + +// loadReportedActor returns the Actor if it is the incarnation actorUID names +// and is on node. Otherwise it returns NotFound, so an atelet learns nothing +// about Actors on other nodes. +func (w *ActorWorkflow) loadReportedActor(ctx context.Context, actorRef resources.ActorRef, actorUID, node string) (*ateapipb.Actor, error) { + actor, err := w.store.GetActor(ctx, actorRef) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return nil, apierror.NotFound("Actor %s not found", actorRef) + } + return nil, err + } + if got := actor.GetMetadata().GetUid(); got != actorUID { + slog.WarnContext(ctx, "Refusing a snapshot state report naming a stale actor incarnation", + slog.Any("actor", actorRef), + slog.String("requested_uid", actorUID), + slog.String("actual_uid", got), + slog.String("caller_node", node)) + return nil, apierror.NotFound("Actor %s not found", actorRef) + } + if !actorOnNode(actor, node) { + slog.WarnContext(ctx, "Refusing a snapshot state report for an actor not on the caller's node", + slog.Any("actor", actorRef), + slog.String("caller_node", node), + slog.String("assigned_node", actor.GetStatus().GetAssignedNode())) + return nil, apierror.NotFound("Actor %s not found", actorRef) + } + return actor, nil +} + +// actorOnNode reports whether the Actor is placed on node or has a local +// snapshot there. Unlike HostingNode, it includes a node that only holds the +// Actor's snapshots, since that node's atelet reports them. +func actorOnNode(actor *ateapipb.Actor, node string) bool { + if actor.GetStatus().GetAssignedNode() == node { + return true + } + for _, snap := range actor.GetStatus().GetSnapshots() { + if snap.GetLocality() == node { + return true + } + } + return false +} + +// applyActorSnapshotState applies the two facts a report carries: an upload +// finished, and the atelet on node detached the Actor. A report with neither +// latest_local nor uploading is a detach. +func applyActorSnapshotState(status *ateapipb.ActorStatus, node string, state *ateapipb.ActorSnapshotState) error { + if err := applyLatestDurable(status, state.GetLatestDurable()); err != nil { + return err + } + if state.GetLatestLocal() == nil && state.GetUploading() == nil { + return applyDetach(status, node, state.GetLatestDurable().GetGeneration()) + } + return nil +} + +// applyLatestDurable marks the reported durable snapshot COMPLETED and records +// the fidelity from its manifest. Its uuid and upload URI must match the +// recorded ones. A generation at or below the latest COMPLETED one changes +// nothing, and so does one status no longer has, such as one a revert +// discarded. +func applyLatestDurable(status *ateapipb.ActorStatus, reported *ateapipb.SnapshotReport) error { + if reported == nil { + return nil + } + if latest := findLatestDurableSnapshot(status, ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED); latest.GetGeneration() >= reported.GetGeneration() { + return nil + } + snap := findSnapshotByGeneration(status.GetSnapshots(), reported.GetGeneration()) + if snap == nil { + if reported.GetGeneration() <= status.GetLastAssignedGeneration() { + return nil + } + return apierror.FailedPrecondition("snapshot generation %d (%s) is not recorded for this actor", reported.GetGeneration(), reported.GetUuid()) + } + if snap.GetUuid() != reported.GetUuid() { + return apierror.FailedPrecondition("snapshot generation %d is recorded as %s, not %s", reported.GetGeneration(), snap.GetUuid(), reported.GetUuid()) + } + want := snap.GetDurableSnapshot().GetObject().GetSnapshotUri() + if want == "" { + return apierror.FailedPrecondition("snapshot generation %d has no upload location recorded", reported.GetGeneration()) + } + if got := reported.GetObject().GetSnapshotUri(); got != want { + return apierror.InvalidArgument("snapshot generation %d was uploaded to %q, want %q", reported.GetGeneration(), got, want) + } + snap.Fidelity = reported.GetFidelity() + snap.DurableSnapshot.Status = ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED + return nil +} + +// applyDetach records that the atelet on node deleted its copies of the +// Actor's snapshots up to generation upTo, the latest snapshot it reported +// durable. A newer snapshot on node postdates the detach, so it stays. A +// SUSPENDED Actor is released from node once none of its snapshots are there. +// It refuses if the latest deleted snapshot isn't durable, so the Actor can't +// resume from an older one. +func applyDetach(status *ateapipb.ActorStatus, node string, upTo int32) error { + deleted := func(s *ateapipb.Snapshot) bool { return s.GetLocality() == node && s.GetGeneration() <= upTo } + if latest := latestSnapshot(status, deleted); latest != nil && + latest.GetDurableSnapshot().GetStatus() != ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED { + return apierror.FailedPrecondition("snapshot generation %d on node %s is not durable", latest.GetGeneration(), node) + } + for _, snap := range status.GetSnapshots() { + if deleted(snap) { + snap.Locality = "" + } + } + onNode := latestSnapshot(status, func(s *ateapipb.Snapshot) bool { return s.GetLocality() == node }) + if status.GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED && status.GetAssignedNode() == node && onNode == nil { + status.AssignedNode = "" + } + return nil +} diff --git a/cmd/ateapi/internal/controlapi/snapshot_report_test.go b/cmd/ateapi/internal/controlapi/snapshot_report_test.go new file mode 100644 index 0000000000..7d7be9b3cf --- /dev/null +++ b/cmd/ateapi/internal/controlapi/snapshot_report_test.go @@ -0,0 +1,439 @@ +// 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" + "fmt" + "testing" + + "github.com/google/go-cmp/cmp" + "google.golang.org/grpc/codes" + "google.golang.org/protobuf/testing/protocmp" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" + "github.com/agent-substrate/substrate/internal/apierror" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +const ( + reportAtespace = "team-a" + reportActorUID = "5f2c1a90-7b34-4e6d-8a11-0c3e9d5b7f42" + reportNode = "node-a" + reportOther = "node-b" +) + +func snapName(gen int32) string { return fmt.Sprintf("snap-%d", gen) } + +// ownedURI returns the upload URI of actorUID's snapshot of generation gen. +func ownedURI(t *testing.T, actorUID string, gen int32) string { + t.Helper() + uri, err := resources.NewActorSnapshotURI("gs://snapshots", reportAtespace, actorUID, snapName(gen)) + if err != nil { + t.Fatal(err) + } + return uri.String() +} + +// localOnly returns a Snapshot held on locality with no durable copy. +func localOnly(gen int32, locality string) *ateapipb.Snapshot { + return &ateapipb.Snapshot{Uuid: snapName(gen), Generation: gen, Locality: locality, Fidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY} +} + +// withDurable returns a Snapshot with a durable copy in the given storage +// status. +func withDurable(t *testing.T, gen int32, locality string, storage ateapipb.SnapshotStorageStatus) *ateapipb.Snapshot { + t.Helper() + snap := localOnly(gen, locality) + snap.DurableSnapshot = &ateapipb.SnapshotStorage{ + Status: storage, + Object: &ateapipb.ObjectSnapshot{SnapshotUri: ownedURI(t, reportActorUID, gen)}, + } + return snap +} + +func reported(t *testing.T, gen int32) *ateapipb.SnapshotReport { + t.Helper() + return reportedAt(t, gen, ownedURI(t, reportActorUID, gen)) +} + +func reportedAt(t *testing.T, gen int32, uri string) *ateapipb.SnapshotReport { + t.Helper() + return &ateapipb.SnapshotReport{ + Uuid: snapName(gen), + Generation: gen, + Fidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, + Object: &ateapipb.ObjectSnapshot{SnapshotUri: uri}, + } +} + +func TestApplyActorSnapshotState(t *testing.T) { + const ( + pending = ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING + completed = ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED + suspended = ateapipb.ActorState_ACTOR_STATE_SUSPENDED + running = ateapipb.ActorState_ACTOR_STATE_RUNNING + ) + tests := []struct { + name string + status *ateapipb.ActorStatus + state *ateapipb.ActorSnapshotState + want *ateapipb.ActorStatus + wantCode codes.Code + }{{ + name: "upload marks the pending snapshot COMPLETED and keeps it local", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, completed), + }}, + }, { + name: "upload records the fidelity the manifest reports", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: func() *ateapipb.SnapshotReport { + r := reported(t, 2) + r.Fidelity = ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_ROOTFS + return r + }()}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + func() *ateapipb.Snapshot { + snap := withDurable(t, 2, reportNode, completed) + snap.Fidelity = ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_ROOTFS + return snap + }(), + }}, + }, { + name: "a snapshot already COMPLETED changes nothing", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, completed), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, completed), + }}, + }, { + name: "a generation below the latest COMPLETED one changes nothing", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, LastAssignedGeneration: 2, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 1, reportNode, pending), + withDurable(t, 2, reportNode, completed), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: func() *ateapipb.SnapshotReport { + r := reported(t, 1) + r.Fidelity = ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_ROOTFS + return r + }()}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, LastAssignedGeneration: 2, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 1, reportNode, pending), + withDurable(t, 2, reportNode, completed), + }}, + }, { + name: "a generation status no longer has changes nothing", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, LastAssignedGeneration: 3, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 3, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 3), LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, LastAssignedGeneration: 3, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 3, reportNode, pending), + }}, + }, { + name: "a generation newer than any assigned is refused", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, LastAssignedGeneration: 2, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: reported(t, 5)}, + wantCode: codes.FailedPrecondition, + }, { + name: "a uuid other than the recorded one is refused", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: func() *ateapipb.SnapshotReport { + r := reported(t, 2) + r.Uuid = "other" + return r + }()}, + wantCode: codes.FailedPrecondition, + }, { + name: "a snapshot with no recorded upload URI is refused", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + localOnly(2, reportNode), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: reported(t, 2)}, + wantCode: codes.FailedPrecondition, + }, { + name: "an upload URI other than the recorded one is refused", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 2), LatestDurable: reportedAt(t, 2, ownedURI(t, reportActorUID, 9))}, + wantCode: codes.InvalidArgument, + }, { + name: "latest_local and uploading are not checked", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestLocal: reported(t, 7), Uploading: reported(t, 8)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + }, { + name: "an upload in flight with no local snapshot is not a detach", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{Uploading: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + }, { + name: "detach releases a suspended actor from the node", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 1, "", completed), + withDurable(t, 2, reportNode, completed), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 1, "", completed), + withDurable(t, 2, "", completed), + }}, + }, { + name: "a detach can complete the upload it reports", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, pending), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, "", completed), + }}, + }, { + name: "a late detach leaves a newer snapshot on the node and keeps the actor placed", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, completed), + withDurable(t, 3, reportNode, completed), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, "", completed), + withDurable(t, 3, reportNode, completed), + }}, + }, { + name: "a late detach leaves a newer local-only snapshot on the node", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 1, "", completed), + localOnly(2, reportNode), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 1)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 1, "", completed), + localOnly(2, reportNode), + }}, + }, { + name: "a detach of a snapshot that isn't durable is refused", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, LastAssignedGeneration: 2, Snapshots: []*ateapipb.Snapshot{ + localOnly(1, reportNode), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 2)}, + wantCode: codes.FailedPrecondition, + }, { + name: "a detach from a node that holds no snapshot only releases the actor", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, "", completed), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, "", completed), + }}, + }, { + name: "a running actor's detach keeps it placed on the node", + status: &ateapipb.ActorStatus{State: running, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, completed), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: running, AssignedNode: reportNode, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, "", completed), + }}, + }, { + name: "a detach from one node leaves another node alone", + status: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportOther, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, reportNode, completed), + localOnly(3, reportOther), + }}, + state: &ateapipb.ActorSnapshotState{LatestDurable: reported(t, 2)}, + want: &ateapipb.ActorStatus{State: suspended, AssignedNode: reportOther, Snapshots: []*ateapipb.Snapshot{ + withDurable(t, 2, "", completed), + localOnly(3, reportOther), + }}, + }} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := applyActorSnapshotState(tt.status, reportNode, tt.state) + if tt.wantCode != codes.OK { + if got := apierror.Code(err); got != tt.wantCode { + t.Fatalf("applyActorSnapshotState() error = %v, want code %v", err, tt.wantCode) + } + return + } + if err != nil { + t.Fatalf("applyActorSnapshotState() error = %v", err) + } + if diff := cmp.Diff(tt.want, tt.status, protocmp.Transform()); diff != "" { + t.Errorf("status mismatch (-want +got):\n%s", diff) + } + }) + } +} + +// A report is applied to the stored actor under its lease. A refused report, +// and a report that changes nothing, write nothing. +func TestApplyReportedSnapshotState(t *testing.T) { + ctx := context.Background() + persistence := newTestPersistence(t) + w := NewActorWorkflow(persistence, nil, nil, nil, nil, nil, "", nil, nil) + tmpl := &ateapipb.ActorTemplate{SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: testStorageLocation}} + + actorRef := resources.ActorRef{Atespace: reportAtespace, Name: "actor-1"} + actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}, + }) + uid := actor.GetMetadata().GetUid() + pending := mustActorSnapshotURI(t, tmpl, actor, "snap-2") + actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { + s.AssignedNode = reportNode + s.LastAssignedGeneration = 2 + snap := newDurableSnapshot(2, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", pending.Name(), pending.String(), ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING) + snap.Locality = reportNode + s.Snapshots = []*ateapipb.Snapshot{snap} + }) + uploaded := reportedAt(t, 2, pending.String()) + state := &ateapipb.ActorSnapshotState{LatestLocal: uploaded, LatestDurable: uploaded} + + assertUnwritten := func(t *testing.T) { + t.Helper() + got, err := persistence.GetActor(ctx, actorRef) + if err != nil { + t.Fatal(err) + } + if got.GetMetadata().GetVersion() != actor.GetMetadata().GetVersion() { + t.Errorf("a refused report wrote the actor: version %d, want %d", got.GetMetadata().GetVersion(), actor.GetMetadata().GetVersion()) + } + } + + t.Run("refused while another operation holds the actor", func(t *testing.T) { + lease, err := persistence.AcquireLease(ctx, actorLeaseKey(actorRef)) + if err != nil { + t.Fatalf("AcquireLease() failed: %v", err) + } + defer lease.Close() + if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, state); apierror.Code(err) != codes.Aborted { + t.Fatalf("ApplyReportedSnapshotState() = %v, want Aborted", err) + } + assertUnwritten(t) + }) + for _, tc := range []struct { + name, uid, node string + ref resources.ActorRef + }{ + {"refused for another node", uid, reportOther, actorRef}, + {"refused for a stale incarnation", "1a7e4c83-6d20-4f95-b3c8-9e0a2f6d4b17", reportNode, actorRef}, + {"refused for a missing actor", uid, reportNode, resources.ActorRef{Atespace: reportAtespace, Name: "no-such-actor"}}, + } { + t.Run(tc.name, func(t *testing.T) { + if err := w.ApplyReportedSnapshotState(ctx, tc.ref, tc.uid, tc.node, state); apierror.Code(err) != codes.NotFound { + t.Fatalf("ApplyReportedSnapshotState() = %v, want NotFound", err) + } + assertUnwritten(t) + }) + } + // A transitioning actor gets Aborted, and a crashed one FailedPrecondition. + for _, tc := range []struct { + state ateapipb.ActorState + want codes.Code + }{ + {ateapipb.ActorState_ACTOR_STATE_SUSPENDING, codes.Aborted}, + {ateapipb.ActorState_ACTOR_STATE_RESUMING, codes.Aborted}, + {ateapipb.ActorState_ACTOR_STATE_DELETING, codes.Aborted}, + {ateapipb.ActorState_ACTOR_STATE_REVERTING, codes.Aborted}, + {ateapipb.ActorState_ACTOR_STATE_CRASHED, codes.FailedPrecondition}, + } { + t.Run("refused for a "+tc.state.String()+" actor", func(t *testing.T) { + actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { s.State = tc.state }) + defer func() { + actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { + s.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDED + }) + }() + if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, state); apierror.Code(err) != tc.want { + t.Fatalf("ApplyReportedSnapshotState() = %v, want %v", err, tc.want) + } + assertUnwritten(t) + }) + } + t.Run("refused for an upload to the wrong URI", func(t *testing.T) { + wrong := &ateapipb.ActorSnapshotState{LatestLocal: uploaded, LatestDurable: reportedAt(t, 2, ownedURI(t, uid, 9))} + if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, wrong); apierror.Code(err) != codes.InvalidArgument { + t.Fatalf("ApplyReportedSnapshotState() = %v, want InvalidArgument", err) + } + assertUnwritten(t) + }) + + if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, state); err != nil { + t.Fatalf("ApplyReportedSnapshotState() failed: %v", err) + } + got, err := persistence.GetActor(ctx, actorRef) + if err != nil { + t.Fatal(err) + } + if st := got.GetStatus().GetSnapshots()[0].GetDurableSnapshot().GetStatus(); st != ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED { + t.Errorf("durable snapshot status = %v, want COMPLETED", st) + } + + // A report that changes nothing writes nothing. + actor = got + if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, state); err != nil { + t.Fatalf("repeated ApplyReportedSnapshotState() failed: %v", err) + } + assertUnwritten(t) +} + +func TestActorOnNode(t *testing.T) { + actor := func(assigned string, localities ...string) *ateapipb.Actor { + status := &ateapipb.ActorStatus{AssignedNode: assigned} + for i, l := range localities { + status.Snapshots = append(status.Snapshots, localOnly(int32(i+1), l)) + } + return &ateapipb.Actor{Status: status} + } + for _, tc := range []struct { + name string + actor *ateapipb.Actor + want bool + }{ + {"placed on the node", actor(reportNode), true}, + {"holds a snapshot on the node", actor(reportOther, "", reportNode), true}, + {"neither", actor(reportOther, reportOther), false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := actorOnNode(tc.actor, reportNode); got != tc.want { + t.Errorf("actorOnNode() = %v, want %v", got, tc.want) + } + }) + } +} From 37ecf8f94d119cc4dbe7124de3e119db8828fab5 Mon Sep 17 00:00:00 2001 From: Ivy Date: Wed, 14 Oct 2026 13:03:40 -0700 Subject: [PATCH 2/2] ateapi: serve WorkerService.ReportActorSnapshotState The handler authenticates atelet before reading the request, since authz doesn't cover WorkerService, validates it, and hands it to the control plane with the node from the caller's certificate. Validation requires a detach, a report with neither latest_local nor uploading, to name the snapshot it made durable in latest_durable. The RPC comment says which two facts the control plane records, that repeating a report is safe, and which errors atelet should retry. --- cmd/ateapi/internal/apivalidation/worker.go | 8 +- .../internal/apivalidation/worker_test.go | 8 +- cmd/ateapi/internal/workerservice/report.go | 43 +++++++ .../internal/workerservice/report_test.go | 106 ++++++++++++++++++ cmd/ateapi/internal/workerservice/service.go | 2 + .../internal/workerservice/service_test.go | 17 +++ pkg/proto/ateapipb/ateapi.proto | 22 ++-- pkg/proto/ateapipb/ateapi_grpc.pb.go | 40 +++---- 8 files changed, 210 insertions(+), 36 deletions(-) create mode 100644 cmd/ateapi/internal/workerservice/report.go create mode 100644 cmd/ateapi/internal/workerservice/report_test.go diff --git a/cmd/ateapi/internal/apivalidation/worker.go b/cmd/ateapi/internal/apivalidation/worker.go index c56056a361..b13769cfe5 100644 --- a/cmd/ateapi/internal/apivalidation/worker.go +++ b/cmd/ateapi/internal/apivalidation/worker.go @@ -96,7 +96,13 @@ func ValidateMarkWorkerUnavailableRequest(ctx context.Context, req *ateapipb.Mar func ValidateReportActorSnapshotStateRequest(ctx context.Context, req *ateapipb.ReportActorSnapshotStateRequest) field.ErrorList { op := operation.Operation{Type: operation.Create} - return Validate_ReportActorSnapshotStateRequest(ctx, op, nil, req, nil) + errs := Validate_ReportActorSnapshotStateRequest(ctx, op, nil, req, nil) + // A report with neither latest_local nor uploading is a detach, which + // names the snapshot it made durable. + if st := req.GetState(); st != nil && st.GetLatestLocal() == nil && st.GetUploading() == nil && st.GetLatestDurable() == nil { + errs = append(errs, field.Required(field.NewPath("state", "latest_durable"), "a report with neither latest_local nor uploading must set latest_durable")) + } + return errs } // validateWorkerUpdate validates a Worker against the previous stored value. diff --git a/cmd/ateapi/internal/apivalidation/worker_test.go b/cmd/ateapi/internal/apivalidation/worker_test.go index 30c246ebdc..13e868fbe3 100644 --- a/cmd/ateapi/internal/apivalidation/worker_test.go +++ b/cmd/ateapi/internal/apivalidation/worker_test.go @@ -697,8 +697,14 @@ func TestValidateReportActorSnapshotStateRequest(t *testing.T) { r.State.UploadDueTime = nil }), }, { - name: "valid empty state", + name: "empty state", req: valid(func(r *ateapipb.ReportActorSnapshotStateRequest) { r.State = &ateapipb.ActorSnapshotState{} }), + want: field.ErrorList{field.Required(field.NewPath("state", "latest_durable"), "")}, + }, { + name: "valid upload in flight with no local snapshot", + req: valid(func(r *ateapipb.ReportActorSnapshotStateRequest) { + r.State = &ateapipb.ActorSnapshotState{Uploading: r.State.GetUploading()} + }), }, { name: "missing actor", req: valid(func(r *ateapipb.ReportActorSnapshotStateRequest) { r.Actor = nil }), diff --git a/cmd/ateapi/internal/workerservice/report.go b/cmd/ateapi/internal/workerservice/report.go new file mode 100644 index 0000000000..3cc6b92d35 --- /dev/null +++ b/cmd/ateapi/internal/workerservice/report.go @@ -0,0 +1,43 @@ +// 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 workerservice + +import ( + "context" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/apivalidation" + "github.com/agent-substrate/substrate/cmd/ateapi/internal/ateletauth" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +// ReportActorSnapshotState records the snapshot state that the calling atelet +// reports for an Actor on its node. +func (s *Server) ReportActorSnapshotState(ctx context.Context, req *ateapipb.ReportActorSnapshotStateRequest) (*ateapipb.ReportActorSnapshotStateResponse, error) { + // Authz doesn't cover WorkerService, so authenticate before reading the + // request. + caller, err := ateletauth.Authenticate(ctx, s.ateletSPIFFEID) + if err != nil { + return nil, err + } + if errs := apivalidation.ValidateReportActorSnapshotStateRequest(ctx, req); len(errs) > 0 { + return nil, resources.ToAPIError(errs) + } + actorRef := resources.ActorRefFromObjectRef(req.GetActor()) + if err := s.suspender.ApplyActorSnapshotState(ctx, actorRef, req.GetActorUid(), caller.NodeName, req.GetState()); err != nil { + return nil, err + } + return &ateapipb.ReportActorSnapshotStateResponse{}, nil +} diff --git a/cmd/ateapi/internal/workerservice/report_test.go b/cmd/ateapi/internal/workerservice/report_test.go new file mode 100644 index 0000000000..398e5517a1 --- /dev/null +++ b/cmd/ateapi/internal/workerservice/report_test.go @@ -0,0 +1,106 @@ +// 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 workerservice + +import ( + "context" + "testing" + + "google.golang.org/grpc/codes" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/ateletauth/ateletauthtest" + "github.com/agent-substrate/substrate/internal/apierror" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +const reportActorUID = "5f2c1a90-7b34-4e6d-8a11-0c3e9d5b7f42" + +func reportRequest() *ateapipb.ReportActorSnapshotStateRequest { + durable := &ateapipb.SnapshotReport{ + Uuid: "snap-1", + Generation: 1, + Fidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, + Object: &ateapipb.ObjectSnapshot{SnapshotUri: "gs://bucket/actors/a/snap-1"}, + LocalCommitTime: timestamppb.Now(), + } + return &ateapipb.ReportActorSnapshotStateRequest{ + Actor: &ateapipb.ObjectRef{Atespace: suspendAtespace, Name: suspendActor}, + ActorUid: reportActorUID, + State: &ateapipb.ActorSnapshotState{LatestDurable: durable}, + } +} + +// A report reaches the control plane with the node from the caller's +// certificate. The handler reads nothing from the store. +func TestReportActorSnapshotState(t *testing.T) { + suspender := &fakeSuspender{} + s := New(nil, suspender, testAteletSPIFFEID, nil, "", nil) + + req := reportRequest() + if _, err := s.ReportActorSnapshotState(ateletauthtest.ContextWith(ateletauthtest.CertOn(t, testNode)), req); err != nil { + t.Fatalf("ReportActorSnapshotState() failed: %v", err) + } + if len(suspender.reports) != 1 { + t.Fatalf("got %d applied reports, want 1", len(suspender.reports)) + } + got := suspender.reports[0] + if want := (resources.ActorRef{Atespace: suspendAtespace, Name: suspendActor}); got.actorRef != want { + t.Errorf("actorRef = %v, want %v", got.actorRef, want) + } + if got.actorUID != reportActorUID || got.node != testNode { + t.Errorf("applied (uid %q, node %q), want (%q, %q)", got.actorUID, got.node, reportActorUID, testNode) + } + if !proto.Equal(got.state, req.GetState()) { + t.Errorf("applied state = %v, want %v", got.state, req.GetState()) + } +} + +// A caller that isn't atelet, or a malformed request, doesn't reach the control +// plane. A refusal from the control plane reaches atelet unchanged. +func TestReportActorSnapshotState_Refused(t *testing.T) { + onNode := ateletauthtest.ContextWith(ateletauthtest.CertOn(t, testNode)) + invalid := reportRequest() + invalid.State = nil + + tests := []struct { + name string + ctx context.Context + req *ateapipb.ReportActorSnapshotStateRequest + controlErr error + want codes.Code + wantApplied bool + }{ + {name: "unauthenticated", ctx: ateletauthtest.ContextWith(nil), req: reportRequest(), want: codes.Unauthenticated}, + {name: "not atelet", ctx: ateletauthtest.ContextWith(ateletauthtest.CertIn(t, "other-namespace", testNode)), req: reportRequest(), want: codes.PermissionDenied}, + {name: "invalid request", ctx: onNode, req: invalid, want: codes.InvalidArgument}, + {name: "control plane refuses", ctx: onNode, req: reportRequest(), controlErr: apierror.NotFound("Actor not found"), want: codes.NotFound, wantApplied: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + suspender := &fakeSuspender{reportErr: tt.controlErr} + s := New(nil, suspender, testAteletSPIFFEID, nil, "", nil) + _, err := s.ReportActorSnapshotState(tt.ctx, tt.req) + if got := apierror.Code(err); got != tt.want { + t.Fatalf("ReportActorSnapshotState() error = %v, want code %v", err, tt.want) + } + if applied := len(suspender.reports) > 0; applied != tt.wantApplied { + t.Errorf("report reached the control plane = %v, want %v", applied, tt.wantApplied) + } + }) + } +} diff --git a/cmd/ateapi/internal/workerservice/service.go b/cmd/ateapi/internal/workerservice/service.go index 435b1113ff..657eedff84 100644 --- a/cmd/ateapi/internal/workerservice/service.go +++ b/cmd/ateapi/internal/workerservice/service.go @@ -22,6 +22,7 @@ import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/localca" "github.com/agent-substrate/substrate/internal/localjwtauthority" + "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) @@ -54,6 +55,7 @@ type Server struct { // drives. *controlapi.RPCService satisfies it. type actorSuspender interface { SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) + ApplyActorSnapshotState(ctx context.Context, actorRef resources.ActorRef, actorUID, node string, state *ateapipb.ActorSnapshotState) error } var _ ateapipb.WorkerServiceServer = (*Server)(nil) diff --git a/cmd/ateapi/internal/workerservice/service_test.go b/cmd/ateapi/internal/workerservice/service_test.go index 8e26beebdc..71455d4dec 100644 --- a/cmd/ateapi/internal/workerservice/service_test.go +++ b/cmd/ateapi/internal/workerservice/service_test.go @@ -20,6 +20,7 @@ import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/installdefaults" + "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) @@ -41,6 +42,9 @@ type fakeSuspender struct { calls []*ateapipb.SuspendActorRequest resp *ateapipb.SuspendActorResponse err error + + reports []appliedReport + reportErr error } func (f *fakeSuspender) SuspendActor(_ context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) { @@ -48,6 +52,19 @@ func (f *fakeSuspender) SuspendActor(_ context.Context, req *ateapipb.SuspendAct return f.resp, f.err } +func (f *fakeSuspender) ApplyActorSnapshotState(_ context.Context, actorRef resources.ActorRef, actorUID, node string, state *ateapipb.ActorSnapshotState) error { + f.reports = append(f.reports, appliedReport{actorRef: actorRef, actorUID: actorUID, node: node, state: state}) + return f.reportErr +} + +// appliedReport is one ApplyActorSnapshotState call. +type appliedReport struct { + actorRef resources.ActorRef + actorUID string + node string + state *ateapipb.ActorSnapshotState +} + // seedReportedWorker registers a Worker on nodeName that has already reported // capacity and its default runtime: identity from the pod, the way the syncer // writes it, plus the result of an earlier registration. A Worker that has diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 233b3fb63e..8e7c492f09 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -2566,18 +2566,16 @@ service WorkerService { // rather than a repeat of its success. rpc RequestActorSuspend(RequestActorSuspendRequest) returns (RequestActorSuspendResponse); - // ReportActorSnapshotState records which snapshots of an Actor the calling - // atelet holds on its node, and which of them are durable. Only atelet can - // observe when an upload finishes or a local copy is deleted, and the - // control plane needs both to decide where the Actor can resume. - // - // Only atelet should call this, with its own client certificate. The - // control plane serves it only for an Actor on the caller's node. - // Idempotent: each report carries the whole state and replaces the previous - // one. Returns ABORTED while another operation holds the Actor; retry. - // - // TODO(#2570): not served yet. Every call returns UNIMPLEMENTED until the - // handler lands. + // ReportActorSnapshotState reports an Actor's snapshot state on the caller's + // node. The control plane records two facts from it: an upload finished, and + // atelet detached the Actor from the node. Only atelet observes either. + // + // atelet calls this with its own client certificate, for an Actor on its + // node. Each report carries the whole state, so repeating one is safe. + // Returns ABORTED while another operation is changing the Actor; retry. + // Returns NOT_FOUND once the Actor is gone or no longer on the caller's + // node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report + // conflicts with the record; don't retry. rpc ReportActorSnapshotState(ReportActorSnapshotStateRequest) returns (ReportActorSnapshotStateResponse); } diff --git a/pkg/proto/ateapipb/ateapi_grpc.pb.go b/pkg/proto/ateapipb/ateapi_grpc.pb.go index 4043f05a5a..68e65e97a1 100644 --- a/pkg/proto/ateapipb/ateapi_grpc.pb.go +++ b/pkg/proto/ateapipb/ateapi_grpc.pb.go @@ -1990,18 +1990,16 @@ type WorkerServiceClient interface { // retries a suspend it had already been granted therefore sees a failure // rather than a repeat of its success. RequestActorSuspend(ctx context.Context, in *RequestActorSuspendRequest, opts ...grpc.CallOption) (*RequestActorSuspendResponse, error) - // ReportActorSnapshotState records which snapshots of an Actor the calling - // atelet holds on its node, and which of them are durable. Only atelet can - // observe when an upload finishes or a local copy is deleted, and the - // control plane needs both to decide where the Actor can resume. + // ReportActorSnapshotState reports an Actor's snapshot state on the caller's + // node. The control plane records two facts from it: an upload finished, and + // atelet detached the Actor from the node. Only atelet observes either. // - // Only atelet should call this, with its own client certificate. The - // control plane serves it only for an Actor on the caller's node. - // Idempotent: each report carries the whole state and replaces the previous - // one. Returns ABORTED while another operation holds the Actor; retry. - // - // TODO(#2570): not served yet. Every call returns UNIMPLEMENTED until the - // handler lands. + // atelet calls this with its own client certificate, for an Actor on its + // node. Each report carries the whole state, so repeating one is safe. + // Returns ABORTED while another operation is changing the Actor; retry. + // Returns NOT_FOUND once the Actor is gone or no longer on the caller's + // node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report + // conflicts with the record; don't retry. ReportActorSnapshotState(ctx context.Context, in *ReportActorSnapshotStateRequest, opts ...grpc.CallOption) (*ReportActorSnapshotStateResponse, error) } @@ -2119,18 +2117,16 @@ type WorkerServiceServer interface { // retries a suspend it had already been granted therefore sees a failure // rather than a repeat of its success. RequestActorSuspend(context.Context, *RequestActorSuspendRequest) (*RequestActorSuspendResponse, error) - // ReportActorSnapshotState records which snapshots of an Actor the calling - // atelet holds on its node, and which of them are durable. Only atelet can - // observe when an upload finishes or a local copy is deleted, and the - // control plane needs both to decide where the Actor can resume. - // - // Only atelet should call this, with its own client certificate. The - // control plane serves it only for an Actor on the caller's node. - // Idempotent: each report carries the whole state and replaces the previous - // one. Returns ABORTED while another operation holds the Actor; retry. + // ReportActorSnapshotState reports an Actor's snapshot state on the caller's + // node. The control plane records two facts from it: an upload finished, and + // atelet detached the Actor from the node. Only atelet observes either. // - // TODO(#2570): not served yet. Every call returns UNIMPLEMENTED until the - // handler lands. + // atelet calls this with its own client certificate, for an Actor on its + // node. Each report carries the whole state, so repeating one is safe. + // Returns ABORTED while another operation is changing the Actor; retry. + // Returns NOT_FOUND once the Actor is gone or no longer on the caller's + // node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report + // conflicts with the record; don't retry. ReportActorSnapshotState(context.Context, *ReportActorSnapshotStateRequest) (*ReportActorSnapshotStateResponse, error) mustEmbedUnimplementedWorkerServiceServer() }