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/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) + } + }) + } +} 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() }