diff --git a/api/ateapi/v1/ateapi.pb.go b/api/ateapi/v1/ateapi.pb.go index d63c010fb1..255d29c465 100644 --- a/api/ateapi/v1/ateapi.pb.go +++ b/api/ateapi/v1/ateapi.pb.go @@ -5245,7 +5245,24 @@ type ResumeActorRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // +k8s:required // +k8s:subfield(atespace)=+k8s:required - Actor *ObjectRef `protobuf:"bytes,1,opt,name=actor,proto3" json:"actor,omitempty"` + Actor *ObjectRef `protobuf:"bytes,1,opt,name=actor,proto3" json:"actor,omitempty"` + // Resume only if this is still the actor and checkpoint the caller observed. + // This guards the current restore source; it does not select an older one. + // Restoring requires SUSPENDED or retryable RESUMING and a matching completed + // durable checkpoint, even when restoring a local copy. If its node was + // deleted, resume attempts that checkpoint's upload even if completion was + // not reported. A missing upload moves the actor to CRASHED and returns + // FAILED_PRECONDITION; call RevertActor before resuming again. + // A RUNNING actor with a matching UID and newest held checkpoint returns + // resumed=false. + // A UID or checkpoint mismatch returns FAILED_PRECONDITION: the caller's + // view is stale, so retrying the same request won't help. An invalid state + // also returns FAILED_PRECONDITION. + // A concurrent template change returns ABORTED and can be retried; a matching + // pending upload on a node not known to be deleted returns UNAVAILABLE. + // Omit for unconditional resume. + // +k8s:optional + Expected *ExpectedSnapshot `protobuf:"bytes,2,opt,name=expected,proto3" json:"expected,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -5287,6 +5304,13 @@ func (x *ResumeActorRequest) GetActor() *ObjectRef { return nil } +func (x *ResumeActorRequest) GetExpected() *ExpectedSnapshot { + if x != nil { + return x.Expected + } + return nil +} + type ResumeActorResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Actor *Actor `protobuf:"bytes,1,opt,name=actor,proto3" json:"actor,omitempty"` @@ -9965,6 +9989,76 @@ func (x *MethodAnnotation) GetResource() string { return "" } +// ExpectedSnapshot identifies the actor incarnation and checkpoint to resume. +type ExpectedSnapshot struct { + state protoimpl.MessageState `protogen:"open.v1"` + // UID of the actor the caller observed. + // +k8s:required + // +k8s:format=k8s-uuid + ActorUid string `protobuf:"bytes,1,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + // UUID of the checkpoint the caller observed. + // +k8s:required + // +k8s:format=k8s-short-name + SnapshotUuid string `protobuf:"bytes,2,opt,name=snapshot_uuid,json=snapshotUuid,proto3" json:"snapshot_uuid,omitempty"` + // Generation of the checkpoint the caller observed. + // +k8s:required + // +k8s:minimum=1 + Generation int32 `protobuf:"varint,3,opt,name=generation,proto3" json:"generation,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExpectedSnapshot) Reset() { + *x = ExpectedSnapshot{} + mi := &file_api_ateapi_v1_ateapi_proto_msgTypes[140] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExpectedSnapshot) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExpectedSnapshot) ProtoMessage() {} + +func (x *ExpectedSnapshot) ProtoReflect() protoreflect.Message { + mi := &file_api_ateapi_v1_ateapi_proto_msgTypes[140] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExpectedSnapshot.ProtoReflect.Descriptor instead. +func (*ExpectedSnapshot) Descriptor() ([]byte, []int) { + return file_api_ateapi_v1_ateapi_proto_rawDescGZIP(), []int{140} +} + +func (x *ExpectedSnapshot) GetActorUid() string { + if x != nil { + return x.ActorUid + } + return "" +} + +func (x *ExpectedSnapshot) GetSnapshotUuid() string { + if x != nil { + return x.SnapshotUuid + } + return "" +} + +func (x *ExpectedSnapshot) GetGeneration() int32 { + if x != nil { + return x.Generation + } + return 0 +} + var file_api_ateapi_v1_ateapi_proto_extTypes = []protoimpl.ExtensionInfo{ { ExtendedType: (*descriptorpb.MessageOptions)(nil), @@ -10252,9 +10346,10 @@ const file_api_ateapi_v1_ateapi_proto_rawDesc = "" + "\x14SuspendActorResponse\x120\n" + "\x05actor\x18\x01 \x01(\v2\x1a.substrate.ateapi.v1.ActorR\x05actor\"I\n" + "\x0eSuspendOptions\x127\n" + - "\x04mode\x18\x01 \x01(\x0e2#.substrate.ateapi.v1.DurabilityModeR\x04mode\"J\n" + + "\x04mode\x18\x01 \x01(\x0e2#.substrate.ateapi.v1.DurabilityModeR\x04mode\"\x8d\x01\n" + "\x12ResumeActorRequest\x124\n" + - "\x05actor\x18\x01 \x01(\v2\x1e.substrate.ateapi.v1.ObjectRefR\x05actor\"a\n" + + "\x05actor\x18\x01 \x01(\v2\x1e.substrate.ateapi.v1.ObjectRefR\x05actor\x12A\n" + + "\bexpected\x18\x02 \x01(\v2%.substrate.ateapi.v1.ExpectedSnapshotR\bexpected\"a\n" + "\x13ResumeActorResponse\x120\n" + "\x05actor\x18\x01 \x01(\v2\x1a.substrate.ateapi.v1.ActorR\x05actor\x12\x18\n" + "\aresumed\x18\x02 \x01(\bR\aresumed\"J\n" + @@ -10498,7 +10593,13 @@ const file_api_ateapi_v1_ateapi_proto_rawDesc = "" + "\x05scope\x18\x03 \x01(\x0e2\x1a.substrate.ateapi.v1.ScopeR\x05scope\x12\x16\n" + "\x06plural\x18\x04 \x01(\tR\x06plural\".\n" + "\x10MethodAnnotation\x12\x1a\n" + - "\bresource\x18\x01 \x01(\tR\bresource*a\n" + + "\bresource\x18\x01 \x01(\tR\bresource\"t\n" + + "\x10ExpectedSnapshot\x12\x1b\n" + + "\tactor_uid\x18\x01 \x01(\tR\bactorUid\x12#\n" + + "\rsnapshot_uuid\x18\x02 \x01(\tR\fsnapshotUuid\x12\x1e\n" + + "\n" + + "generation\x18\x03 \x01(\x05R\n" + + "generation*a\n" + "\rSnapshotOwner\x12\x1e\n" + "\x1aSNAPSHOT_OWNER_UNSPECIFIED\x10\x00\x12\x18\n" + "\x14SNAPSHOT_OWNER_ACTOR\x10\x01\x12\x16\n" + @@ -10663,7 +10764,7 @@ func file_api_ateapi_v1_ateapi_proto_rawDescGZIP() []byte { } var file_api_ateapi_v1_ateapi_proto_enumTypes = make([]protoimpl.EnumInfo, 10) -var file_api_ateapi_v1_ateapi_proto_msgTypes = make([]protoimpl.MessageInfo, 143) +var file_api_ateapi_v1_ateapi_proto_msgTypes = make([]protoimpl.MessageInfo, 144) var file_api_ateapi_v1_ateapi_proto_goTypes = []any{ (SnapshotOwner)(0), // 0: substrate.ateapi.v1.SnapshotOwner (SnapshotStorageState)(0), // 1: substrate.ateapi.v1.SnapshotStorageState @@ -10815,12 +10916,13 @@ var file_api_ateapi_v1_ateapi_proto_goTypes = []any{ (*DeleteAtespaceAccessPolicyRequest)(nil), // 147: substrate.ateapi.v1.DeleteAtespaceAccessPolicyRequest (*ResourceAnnotation)(nil), // 148: substrate.ateapi.v1.ResourceAnnotation (*MethodAnnotation)(nil), // 149: substrate.ateapi.v1.MethodAnnotation - nil, // 150: substrate.ateapi.v1.Selector.MatchLabelsEntry - nil, // 151: substrate.ateapi.v1.ExternalVolumeStatus.VolumeContextEntry - nil, // 152: substrate.ateapi.v1.Worker.LabelsEntry - (*timestamppb.Timestamp)(nil), // 153: google.protobuf.Timestamp - (*descriptorpb.MessageOptions)(nil), // 154: google.protobuf.MessageOptions - (*descriptorpb.MethodOptions)(nil), // 155: google.protobuf.MethodOptions + (*ExpectedSnapshot)(nil), // 150: substrate.ateapi.v1.ExpectedSnapshot + nil, // 151: substrate.ateapi.v1.Selector.MatchLabelsEntry + nil, // 152: substrate.ateapi.v1.ExternalVolumeStatus.VolumeContextEntry + nil, // 153: substrate.ateapi.v1.Worker.LabelsEntry + (*timestamppb.Timestamp)(nil), // 154: google.protobuf.Timestamp + (*descriptorpb.MessageOptions)(nil), // 155: google.protobuf.MessageOptions + (*descriptorpb.MethodOptions)(nil), // 156: google.protobuf.MethodOptions } var file_api_ateapi_v1_ateapi_proto_depIdxs = []int32{ 1, // 0: substrate.ateapi.v1.SnapshotStorage.state:type_name -> substrate.ateapi.v1.SnapshotStorageState @@ -10829,12 +10931,12 @@ var file_api_ateapi_v1_ateapi_proto_depIdxs = []int32{ 2, // 3: substrate.ateapi.v1.Snapshot.fidelity:type_name -> substrate.ateapi.v1.SnapshotFidelity 11, // 4: substrate.ateapi.v1.Snapshot.durable_snapshot:type_name -> substrate.ateapi.v1.SnapshotStorage 116, // 5: substrate.ateapi.v1.Snapshot.sandbox_runtime:type_name -> substrate.ateapi.v1.ActorSandboxRuntime - 150, // 6: substrate.ateapi.v1.Selector.match_labels:type_name -> substrate.ateapi.v1.Selector.MatchLabelsEntry - 153, // 7: substrate.ateapi.v1.ResourceMetadata.create_time:type_name -> google.protobuf.Timestamp - 153, // 8: substrate.ateapi.v1.ResourceMetadata.update_time:type_name -> google.protobuf.Timestamp - 153, // 9: substrate.ateapi.v1.ResourceMetadata.delete_time:type_name -> google.protobuf.Timestamp + 151, // 6: substrate.ateapi.v1.Selector.match_labels:type_name -> substrate.ateapi.v1.Selector.MatchLabelsEntry + 154, // 7: substrate.ateapi.v1.ResourceMetadata.create_time:type_name -> google.protobuf.Timestamp + 154, // 8: substrate.ateapi.v1.ResourceMetadata.update_time:type_name -> google.protobuf.Timestamp + 154, // 9: substrate.ateapi.v1.ResourceMetadata.delete_time:type_name -> google.protobuf.Timestamp 16, // 10: substrate.ateapi.v1.ExternalVolumes.volumes:type_name -> substrate.ateapi.v1.ExternalVolumeStatus - 151, // 11: substrate.ateapi.v1.ExternalVolumeStatus.volume_context:type_name -> substrate.ateapi.v1.ExternalVolumeStatus.VolumeContextEntry + 152, // 11: substrate.ateapi.v1.ExternalVolumeStatus.volume_context:type_name -> substrate.ateapi.v1.ExternalVolumeStatus.VolumeContextEntry 14, // 12: substrate.ateapi.v1.Actor.metadata:type_name -> substrate.ateapi.v1.ResourceMetadata 35, // 13: substrate.ateapi.v1.Actor.actor_template:type_name -> substrate.ateapi.v1.ObjectRef 13, // 14: substrate.ateapi.v1.Actor.worker_selector:type_name -> substrate.ateapi.v1.Selector @@ -10859,7 +10961,7 @@ var file_api_ateapi_v1_ateapi_proto_depIdxs = []int32{ 15, // 33: substrate.ateapi.v1.ActorStatus.external_volumes:type_name -> substrate.ateapi.v1.ExternalVolumes 30, // 34: substrate.ateapi.v1.ActorStatus.crash:type_name -> substrate.ateapi.v1.ActorCrash 12, // 35: substrate.ateapi.v1.ActorStatus.snapshots:type_name -> substrate.ateapi.v1.Snapshot - 153, // 36: substrate.ateapi.v1.ActorCrash.crash_time:type_name -> google.protobuf.Timestamp + 154, // 36: substrate.ateapi.v1.ActorCrash.crash_time:type_name -> google.protobuf.Timestamp 35, // 37: substrate.ateapi.v1.WorkerAssignment.worker:type_name -> substrate.ateapi.v1.ObjectRef 116, // 38: substrate.ateapi.v1.WorkerAssignment.sandbox_runtime:type_name -> substrate.ateapi.v1.ActorSandboxRuntime 12, // 39: substrate.ateapi.v1.TagStatus.snapshot:type_name -> substrate.ateapi.v1.Snapshot @@ -10878,7 +10980,7 @@ var file_api_ateapi_v1_ateapi_proto_depIdxs = []int32{ 40, // 52: substrate.ateapi.v1.ActorTemplate.status:type_name -> substrate.ateapi.v1.ActorTemplateStatus 38, // 53: substrate.ateapi.v1.ResourceRequirements.limits:type_name -> substrate.ateapi.v1.Resources 35, // 54: substrate.ateapi.v1.GoldenSnapshotStatus.golden_tag:type_name -> substrate.ateapi.v1.ObjectRef - 153, // 55: substrate.ateapi.v1.GoldenSnapshotStatus.snapshot_time:type_name -> google.protobuf.Timestamp + 154, // 55: substrate.ateapi.v1.GoldenSnapshotStatus.snapshot_time:type_name -> google.protobuf.Timestamp 39, // 56: substrate.ateapi.v1.ActorTemplateStatus.golden_snapshot_status:type_name -> substrate.ateapi.v1.GoldenSnapshotStatus 2, // 57: substrate.ateapi.v1.SnapshotConfig.preferred_fidelity:type_name -> substrate.ateapi.v1.SnapshotFidelity 42, // 58: substrate.ateapi.v1.SnapshotConfig.golden_snapshot_config:type_name -> substrate.ateapi.v1.GoldenSnapshotConfig @@ -10920,209 +11022,210 @@ var file_api_ateapi_v1_ateapi_proto_depIdxs = []int32{ 17, // 94: substrate.ateapi.v1.SuspendActorResponse.actor:type_name -> substrate.ateapi.v1.Actor 7, // 95: substrate.ateapi.v1.SuspendOptions.mode:type_name -> substrate.ateapi.v1.DurabilityMode 35, // 96: substrate.ateapi.v1.ResumeActorRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 17, // 97: substrate.ateapi.v1.ResumeActorResponse.actor:type_name -> substrate.ateapi.v1.Actor - 35, // 98: substrate.ateapi.v1.RevertActorRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 17, // 99: substrate.ateapi.v1.RevertActorResponse.actor:type_name -> substrate.ateapi.v1.Actor - 35, // 100: substrate.ateapi.v1.DeleteActorRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 93, // 101: substrate.ateapi.v1.DeleteActorRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions - 35, // 102: substrate.ateapi.v1.GetActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 103: substrate.ateapi.v1.CreateActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 18, // 104: substrate.ateapi.v1.CreateActorEgressPolicyRequest.egress_policy:type_name -> substrate.ateapi.v1.EgressPolicy - 35, // 105: substrate.ateapi.v1.UpdateActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 18, // 106: substrate.ateapi.v1.UpdateActorEgressPolicyRequest.egress_policy:type_name -> substrate.ateapi.v1.EgressPolicy - 35, // 107: substrate.ateapi.v1.DeleteActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 93, // 108: substrate.ateapi.v1.DeleteActorEgressPolicyRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions - 35, // 109: substrate.ateapi.v1.GetTagRequest.tag:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 110: substrate.ateapi.v1.MintActorJwtRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 153, // 111: substrate.ateapi.v1.MintActorJwtResponse.expire_time:type_name -> google.protobuf.Timestamp - 94, // 112: substrate.ateapi.v1.ListTagsRequest.options:type_name -> substrate.ateapi.v1.ListOptions - 33, // 113: substrate.ateapi.v1.ListTagsResponse.tags:type_name -> substrate.ateapi.v1.Tag - 33, // 114: substrate.ateapi.v1.CreateTagRequest.tag:type_name -> substrate.ateapi.v1.Tag - 33, // 115: substrate.ateapi.v1.UpdateTagRequest.tag:type_name -> substrate.ateapi.v1.Tag - 35, // 116: substrate.ateapi.v1.DeleteTagRequest.tag:type_name -> substrate.ateapi.v1.ObjectRef - 93, // 117: substrate.ateapi.v1.DeleteTagRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions - 35, // 118: substrate.ateapi.v1.ListWorkerActorAssignmentsRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 94, // 119: substrate.ateapi.v1.ListWorkerActorAssignmentsRequest.options:type_name -> substrate.ateapi.v1.ListOptions - 120, // 120: substrate.ateapi.v1.ListWorkerActorAssignmentsResponse.actor_assignments:type_name -> substrate.ateapi.v1.ActorAssignment - 94, // 121: substrate.ateapi.v1.ListWorkersRequest.options:type_name -> substrate.ateapi.v1.ListOptions - 111, // 122: substrate.ateapi.v1.ListWorkersResponse.workers:type_name -> substrate.ateapi.v1.Worker - 35, // 123: substrate.ateapi.v1.GetWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 111, // 124: substrate.ateapi.v1.CreateWorkerRequest.worker:type_name -> substrate.ateapi.v1.Worker - 111, // 125: substrate.ateapi.v1.UpdateWorkerRequest.worker:type_name -> substrate.ateapi.v1.Worker - 35, // 126: substrate.ateapi.v1.DeleteWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 93, // 127: substrate.ateapi.v1.DeleteWorkerRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions - 35, // 128: substrate.ateapi.v1.DrainWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 111, // 129: substrate.ateapi.v1.DrainWorkerResponse.worker:type_name -> substrate.ateapi.v1.Worker - 35, // 130: substrate.ateapi.v1.MarkWorkerAvailableRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 111, // 131: substrate.ateapi.v1.MarkWorkerAvailableResponse.worker:type_name -> substrate.ateapi.v1.Worker - 35, // 132: substrate.ateapi.v1.MarkWorkerUnavailableRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 111, // 133: substrate.ateapi.v1.MarkWorkerUnavailableResponse.worker:type_name -> substrate.ateapi.v1.Worker - 94, // 134: substrate.ateapi.v1.ListActorsRequest.options:type_name -> substrate.ateapi.v1.ListOptions - 17, // 135: substrate.ateapi.v1.ListActorsResponse.actors:type_name -> substrate.ateapi.v1.Actor - 14, // 136: substrate.ateapi.v1.Worker.metadata:type_name -> substrate.ateapi.v1.ResourceMetadata - 152, // 137: substrate.ateapi.v1.Worker.labels:type_name -> substrate.ateapi.v1.Worker.LabelsEntry - 112, // 138: substrate.ateapi.v1.Worker.status:type_name -> substrate.ateapi.v1.WorkerStatus - 8, // 139: substrate.ateapi.v1.WorkerStatus.state:type_name -> substrate.ateapi.v1.WorkerState - 119, // 140: substrate.ateapi.v1.WorkerStatus.capacity:type_name -> substrate.ateapi.v1.WorkerResources - 119, // 141: substrate.ateapi.v1.WorkerStatus.allocated:type_name -> substrate.ateapi.v1.WorkerResources - 113, // 142: substrate.ateapi.v1.WorkerStatus.sandbox_classes:type_name -> substrate.ateapi.v1.WorkerSandboxClass - 114, // 143: substrate.ateapi.v1.WorkerSandboxClass.default_runtime:type_name -> substrate.ateapi.v1.WorkerSandboxRuntime - 114, // 144: substrate.ateapi.v1.WorkerSandboxClass.restorable_runtimes:type_name -> substrate.ateapi.v1.WorkerSandboxRuntime - 117, // 145: substrate.ateapi.v1.WorkerSandboxRuntime.version:type_name -> substrate.ateapi.v1.VersionedSandboxCompat - 115, // 146: substrate.ateapi.v1.WorkerSandboxRuntime.actor_overhead:type_name -> substrate.ateapi.v1.RuntimeOverhead - 117, // 147: substrate.ateapi.v1.ActorSandboxRuntime.version:type_name -> substrate.ateapi.v1.VersionedSandboxCompat - 118, // 148: substrate.ateapi.v1.VersionedSandboxCompat.attributes:type_name -> substrate.ateapi.v1.AttributeEntry - 14, // 149: substrate.ateapi.v1.ActorAssignment.metadata:type_name -> substrate.ateapi.v1.ResourceMetadata - 35, // 150: substrate.ateapi.v1.ActorAssignment.actor:type_name -> substrate.ateapi.v1.ObjectRef - 38, // 151: substrate.ateapi.v1.ActorAssignment.resources:type_name -> substrate.ateapi.v1.Resources - 35, // 152: substrate.ateapi.v1.AcquireWorkerEpochRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 153: substrate.ateapi.v1.RegisterWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 119, // 154: substrate.ateapi.v1.RegisterWorkerRequest.capacity:type_name -> substrate.ateapi.v1.WorkerResources - 111, // 155: substrate.ateapi.v1.RegisterWorkerResponse.worker:type_name -> substrate.ateapi.v1.Worker - 35, // 156: substrate.ateapi.v1.SetWorkerSandboxRuntimesRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 113, // 157: substrate.ateapi.v1.SetWorkerSandboxRuntimesRequest.sandbox_classes:type_name -> substrate.ateapi.v1.WorkerSandboxClass - 111, // 158: substrate.ateapi.v1.SetWorkerSandboxRuntimesResponse.worker:type_name -> substrate.ateapi.v1.Worker - 35, // 159: substrate.ateapi.v1.MintWorkerActorCertificateRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 160: substrate.ateapi.v1.MintWorkerActorCertificateRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 161: substrate.ateapi.v1.MintWorkerActorJwtRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 153, // 162: substrate.ateapi.v1.MintWorkerActorJwtResponse.expire_time:type_name -> google.protobuf.Timestamp - 35, // 163: substrate.ateapi.v1.RequestActorSuspendRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 164: substrate.ateapi.v1.RequestActorSuspendRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 17, // 165: substrate.ateapi.v1.RequestActorSuspendResponse.actor:type_name -> substrate.ateapi.v1.Actor - 35, // 166: substrate.ateapi.v1.ReportActorCrashedRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 167: substrate.ateapi.v1.ReportActorCrashedRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 17, // 168: substrate.ateapi.v1.ReportActorCrashedResponse.actor:type_name -> substrate.ateapi.v1.Actor - 35, // 169: substrate.ateapi.v1.ReportActorSnapshotStateRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef - 137, // 170: substrate.ateapi.v1.ReportActorSnapshotStateRequest.state:type_name -> substrate.ateapi.v1.ActorSnapshotState - 138, // 171: substrate.ateapi.v1.ActorSnapshotState.latest_local:type_name -> substrate.ateapi.v1.SnapshotReport - 138, // 172: substrate.ateapi.v1.ActorSnapshotState.uploading:type_name -> substrate.ateapi.v1.SnapshotReport - 138, // 173: substrate.ateapi.v1.ActorSnapshotState.latest_durable:type_name -> substrate.ateapi.v1.SnapshotReport - 153, // 174: substrate.ateapi.v1.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp - 2, // 175: substrate.ateapi.v1.SnapshotReport.fidelity:type_name -> substrate.ateapi.v1.SnapshotFidelity - 10, // 176: substrate.ateapi.v1.SnapshotReport.object:type_name -> substrate.ateapi.v1.ObjectSnapshot - 153, // 177: substrate.ateapi.v1.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp - 153, // 178: substrate.ateapi.v1.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp - 14, // 179: substrate.ateapi.v1.AccessPolicy.metadata:type_name -> substrate.ateapi.v1.ResourceMetadata - 140, // 180: substrate.ateapi.v1.AccessPolicy.bindings:type_name -> substrate.ateapi.v1.Binding - 139, // 181: substrate.ateapi.v1.CreateGlobalAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy - 139, // 182: substrate.ateapi.v1.UpdateGlobalAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy - 35, // 183: substrate.ateapi.v1.GetAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef - 35, // 184: substrate.ateapi.v1.CreateAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef - 139, // 185: substrate.ateapi.v1.CreateAtespaceAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy - 35, // 186: substrate.ateapi.v1.UpdateAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef - 139, // 187: substrate.ateapi.v1.UpdateAtespaceAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy - 35, // 188: substrate.ateapi.v1.DeleteAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef - 93, // 189: substrate.ateapi.v1.DeleteAtespaceAccessPolicyRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions - 9, // 190: substrate.ateapi.v1.ResourceAnnotation.scope:type_name -> substrate.ateapi.v1.Scope - 154, // 191: substrate.ateapi.v1.resource:extendee -> google.protobuf.MessageOptions - 155, // 192: substrate.ateapi.v1.method:extendee -> google.protobuf.MethodOptions - 148, // 193: substrate.ateapi.v1.resource:type_name -> substrate.ateapi.v1.ResourceAnnotation - 149, // 194: substrate.ateapi.v1.method:type_name -> substrate.ateapi.v1.MethodAnnotation - 70, // 195: substrate.ateapi.v1.Control.GetActor:input_type -> substrate.ateapi.v1.GetActorRequest - 71, // 196: substrate.ateapi.v1.Control.CreateActor:input_type -> substrate.ateapi.v1.CreateActorRequest - 72, // 197: substrate.ateapi.v1.Control.UpdateActor:input_type -> substrate.ateapi.v1.UpdateActorRequest - 73, // 198: substrate.ateapi.v1.Control.SuspendActor:input_type -> substrate.ateapi.v1.SuspendActorRequest - 76, // 199: substrate.ateapi.v1.Control.ResumeActor:input_type -> substrate.ateapi.v1.ResumeActorRequest - 78, // 200: substrate.ateapi.v1.Control.RevertActor:input_type -> substrate.ateapi.v1.RevertActorRequest - 80, // 201: substrate.ateapi.v1.Control.DeleteActor:input_type -> substrate.ateapi.v1.DeleteActorRequest - 81, // 202: substrate.ateapi.v1.Control.GetActorEgressPolicy:input_type -> substrate.ateapi.v1.GetActorEgressPolicyRequest - 82, // 203: substrate.ateapi.v1.Control.CreateActorEgressPolicy:input_type -> substrate.ateapi.v1.CreateActorEgressPolicyRequest - 83, // 204: substrate.ateapi.v1.Control.UpdateActorEgressPolicy:input_type -> substrate.ateapi.v1.UpdateActorEgressPolicyRequest - 84, // 205: substrate.ateapi.v1.Control.DeleteActorEgressPolicy:input_type -> substrate.ateapi.v1.DeleteActorEgressPolicyRequest - 86, // 206: substrate.ateapi.v1.Control.MintActorJwt:input_type -> substrate.ateapi.v1.MintActorJwtRequest - 90, // 207: substrate.ateapi.v1.Control.CreateTag:input_type -> substrate.ateapi.v1.CreateTagRequest - 85, // 208: substrate.ateapi.v1.Control.GetTag:input_type -> substrate.ateapi.v1.GetTagRequest - 88, // 209: substrate.ateapi.v1.Control.ListTags:input_type -> substrate.ateapi.v1.ListTagsRequest - 91, // 210: substrate.ateapi.v1.Control.UpdateTag:input_type -> substrate.ateapi.v1.UpdateTagRequest - 92, // 211: substrate.ateapi.v1.Control.DeleteTag:input_type -> substrate.ateapi.v1.DeleteTagRequest - 97, // 212: substrate.ateapi.v1.Control.ListWorkers:input_type -> substrate.ateapi.v1.ListWorkersRequest - 99, // 213: substrate.ateapi.v1.Control.GetWorker:input_type -> substrate.ateapi.v1.GetWorkerRequest - 100, // 214: substrate.ateapi.v1.Control.CreateWorker:input_type -> substrate.ateapi.v1.CreateWorkerRequest - 101, // 215: substrate.ateapi.v1.Control.UpdateWorker:input_type -> substrate.ateapi.v1.UpdateWorkerRequest - 102, // 216: substrate.ateapi.v1.Control.DeleteWorker:input_type -> substrate.ateapi.v1.DeleteWorkerRequest - 103, // 217: substrate.ateapi.v1.Control.DrainWorker:input_type -> substrate.ateapi.v1.DrainWorkerRequest - 105, // 218: substrate.ateapi.v1.Control.MarkWorkerAvailable:input_type -> substrate.ateapi.v1.MarkWorkerAvailableRequest - 107, // 219: substrate.ateapi.v1.Control.MarkWorkerUnavailable:input_type -> substrate.ateapi.v1.MarkWorkerUnavailableRequest - 95, // 220: substrate.ateapi.v1.Control.ListWorkerActorAssignments:input_type -> substrate.ateapi.v1.ListWorkerActorAssignmentsRequest - 109, // 221: substrate.ateapi.v1.Control.ListActors:input_type -> substrate.ateapi.v1.ListActorsRequest - 60, // 222: substrate.ateapi.v1.Control.CreateAtespace:input_type -> substrate.ateapi.v1.CreateAtespaceRequest - 61, // 223: substrate.ateapi.v1.Control.GetAtespace:input_type -> substrate.ateapi.v1.GetAtespaceRequest - 62, // 224: substrate.ateapi.v1.Control.ListAtespaces:input_type -> substrate.ateapi.v1.ListAtespacesRequest - 64, // 225: substrate.ateapi.v1.Control.DeleteAtespace:input_type -> substrate.ateapi.v1.DeleteAtespaceRequest - 65, // 226: substrate.ateapi.v1.Control.CreateActorTemplate:input_type -> substrate.ateapi.v1.CreateActorTemplateRequest - 66, // 227: substrate.ateapi.v1.Control.GetActorTemplate:input_type -> substrate.ateapi.v1.GetActorTemplateRequest - 67, // 228: substrate.ateapi.v1.Control.ListActorTemplates:input_type -> substrate.ateapi.v1.ListActorTemplatesRequest - 69, // 229: substrate.ateapi.v1.Control.DeleteActorTemplate:input_type -> substrate.ateapi.v1.DeleteActorTemplateRequest - 141, // 230: substrate.ateapi.v1.Control.GetGlobalAccessPolicy:input_type -> substrate.ateapi.v1.GetGlobalAccessPolicyRequest - 142, // 231: substrate.ateapi.v1.Control.CreateGlobalAccessPolicy:input_type -> substrate.ateapi.v1.CreateGlobalAccessPolicyRequest - 143, // 232: substrate.ateapi.v1.Control.UpdateGlobalAccessPolicy:input_type -> substrate.ateapi.v1.UpdateGlobalAccessPolicyRequest - 144, // 233: substrate.ateapi.v1.Control.GetAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.GetAtespaceAccessPolicyRequest - 145, // 234: substrate.ateapi.v1.Control.CreateAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.CreateAtespaceAccessPolicyRequest - 146, // 235: substrate.ateapi.v1.Control.UpdateAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.UpdateAtespaceAccessPolicyRequest - 147, // 236: substrate.ateapi.v1.Control.DeleteAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.DeleteAtespaceAccessPolicyRequest - 121, // 237: substrate.ateapi.v1.WorkerService.AcquireWorkerEpoch:input_type -> substrate.ateapi.v1.AcquireWorkerEpochRequest - 123, // 238: substrate.ateapi.v1.WorkerService.RegisterWorker:input_type -> substrate.ateapi.v1.RegisterWorkerRequest - 125, // 239: substrate.ateapi.v1.WorkerService.SetWorkerSandboxRuntimes:input_type -> substrate.ateapi.v1.SetWorkerSandboxRuntimesRequest - 127, // 240: substrate.ateapi.v1.WorkerService.MintWorkerActorCertificate:input_type -> substrate.ateapi.v1.MintWorkerActorCertificateRequest - 129, // 241: substrate.ateapi.v1.WorkerService.MintWorkerActorJwt:input_type -> substrate.ateapi.v1.MintWorkerActorJwtRequest - 131, // 242: substrate.ateapi.v1.WorkerService.RequestActorSuspend:input_type -> substrate.ateapi.v1.RequestActorSuspendRequest - 135, // 243: substrate.ateapi.v1.WorkerService.ReportActorSnapshotState:input_type -> substrate.ateapi.v1.ReportActorSnapshotStateRequest - 133, // 244: substrate.ateapi.v1.WorkerService.ReportActorCrashed:input_type -> substrate.ateapi.v1.ReportActorCrashedRequest - 17, // 245: substrate.ateapi.v1.Control.GetActor:output_type -> substrate.ateapi.v1.Actor - 17, // 246: substrate.ateapi.v1.Control.CreateActor:output_type -> substrate.ateapi.v1.Actor - 17, // 247: substrate.ateapi.v1.Control.UpdateActor:output_type -> substrate.ateapi.v1.Actor - 74, // 248: substrate.ateapi.v1.Control.SuspendActor:output_type -> substrate.ateapi.v1.SuspendActorResponse - 77, // 249: substrate.ateapi.v1.Control.ResumeActor:output_type -> substrate.ateapi.v1.ResumeActorResponse - 79, // 250: substrate.ateapi.v1.Control.RevertActor:output_type -> substrate.ateapi.v1.RevertActorResponse - 17, // 251: substrate.ateapi.v1.Control.DeleteActor:output_type -> substrate.ateapi.v1.Actor - 18, // 252: substrate.ateapi.v1.Control.GetActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy - 18, // 253: substrate.ateapi.v1.Control.CreateActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy - 18, // 254: substrate.ateapi.v1.Control.UpdateActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy - 18, // 255: substrate.ateapi.v1.Control.DeleteActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy - 87, // 256: substrate.ateapi.v1.Control.MintActorJwt:output_type -> substrate.ateapi.v1.MintActorJwtResponse - 33, // 257: substrate.ateapi.v1.Control.CreateTag:output_type -> substrate.ateapi.v1.Tag - 33, // 258: substrate.ateapi.v1.Control.GetTag:output_type -> substrate.ateapi.v1.Tag - 89, // 259: substrate.ateapi.v1.Control.ListTags:output_type -> substrate.ateapi.v1.ListTagsResponse - 33, // 260: substrate.ateapi.v1.Control.UpdateTag:output_type -> substrate.ateapi.v1.Tag - 33, // 261: substrate.ateapi.v1.Control.DeleteTag:output_type -> substrate.ateapi.v1.Tag - 98, // 262: substrate.ateapi.v1.Control.ListWorkers:output_type -> substrate.ateapi.v1.ListWorkersResponse - 111, // 263: substrate.ateapi.v1.Control.GetWorker:output_type -> substrate.ateapi.v1.Worker - 111, // 264: substrate.ateapi.v1.Control.CreateWorker:output_type -> substrate.ateapi.v1.Worker - 111, // 265: substrate.ateapi.v1.Control.UpdateWorker:output_type -> substrate.ateapi.v1.Worker - 111, // 266: substrate.ateapi.v1.Control.DeleteWorker:output_type -> substrate.ateapi.v1.Worker - 104, // 267: substrate.ateapi.v1.Control.DrainWorker:output_type -> substrate.ateapi.v1.DrainWorkerResponse - 106, // 268: substrate.ateapi.v1.Control.MarkWorkerAvailable:output_type -> substrate.ateapi.v1.MarkWorkerAvailableResponse - 108, // 269: substrate.ateapi.v1.Control.MarkWorkerUnavailable:output_type -> substrate.ateapi.v1.MarkWorkerUnavailableResponse - 96, // 270: substrate.ateapi.v1.Control.ListWorkerActorAssignments:output_type -> substrate.ateapi.v1.ListWorkerActorAssignmentsResponse - 110, // 271: substrate.ateapi.v1.Control.ListActors:output_type -> substrate.ateapi.v1.ListActorsResponse - 34, // 272: substrate.ateapi.v1.Control.CreateAtespace:output_type -> substrate.ateapi.v1.Atespace - 34, // 273: substrate.ateapi.v1.Control.GetAtespace:output_type -> substrate.ateapi.v1.Atespace - 63, // 274: substrate.ateapi.v1.Control.ListAtespaces:output_type -> substrate.ateapi.v1.ListAtespacesResponse - 34, // 275: substrate.ateapi.v1.Control.DeleteAtespace:output_type -> substrate.ateapi.v1.Atespace - 36, // 276: substrate.ateapi.v1.Control.CreateActorTemplate:output_type -> substrate.ateapi.v1.ActorTemplate - 36, // 277: substrate.ateapi.v1.Control.GetActorTemplate:output_type -> substrate.ateapi.v1.ActorTemplate - 68, // 278: substrate.ateapi.v1.Control.ListActorTemplates:output_type -> substrate.ateapi.v1.ListActorTemplatesResponse - 36, // 279: substrate.ateapi.v1.Control.DeleteActorTemplate:output_type -> substrate.ateapi.v1.ActorTemplate - 139, // 280: substrate.ateapi.v1.Control.GetGlobalAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy - 139, // 281: substrate.ateapi.v1.Control.CreateGlobalAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy - 139, // 282: substrate.ateapi.v1.Control.UpdateGlobalAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy - 139, // 283: substrate.ateapi.v1.Control.GetAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy - 139, // 284: substrate.ateapi.v1.Control.CreateAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy - 139, // 285: substrate.ateapi.v1.Control.UpdateAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy - 139, // 286: substrate.ateapi.v1.Control.DeleteAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy - 122, // 287: substrate.ateapi.v1.WorkerService.AcquireWorkerEpoch:output_type -> substrate.ateapi.v1.AcquireWorkerEpochResponse - 124, // 288: substrate.ateapi.v1.WorkerService.RegisterWorker:output_type -> substrate.ateapi.v1.RegisterWorkerResponse - 126, // 289: substrate.ateapi.v1.WorkerService.SetWorkerSandboxRuntimes:output_type -> substrate.ateapi.v1.SetWorkerSandboxRuntimesResponse - 128, // 290: substrate.ateapi.v1.WorkerService.MintWorkerActorCertificate:output_type -> substrate.ateapi.v1.MintWorkerActorCertificateResponse - 130, // 291: substrate.ateapi.v1.WorkerService.MintWorkerActorJwt:output_type -> substrate.ateapi.v1.MintWorkerActorJwtResponse - 132, // 292: substrate.ateapi.v1.WorkerService.RequestActorSuspend:output_type -> substrate.ateapi.v1.RequestActorSuspendResponse - 136, // 293: substrate.ateapi.v1.WorkerService.ReportActorSnapshotState:output_type -> substrate.ateapi.v1.ReportActorSnapshotStateResponse - 134, // 294: substrate.ateapi.v1.WorkerService.ReportActorCrashed:output_type -> substrate.ateapi.v1.ReportActorCrashedResponse - 245, // [245:295] is the sub-list for method output_type - 195, // [195:245] is the sub-list for method input_type - 193, // [193:195] is the sub-list for extension type_name - 191, // [191:193] is the sub-list for extension extendee - 0, // [0:191] is the sub-list for field type_name + 150, // 97: substrate.ateapi.v1.ResumeActorRequest.expected:type_name -> substrate.ateapi.v1.ExpectedSnapshot + 17, // 98: substrate.ateapi.v1.ResumeActorResponse.actor:type_name -> substrate.ateapi.v1.Actor + 35, // 99: substrate.ateapi.v1.RevertActorRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 17, // 100: substrate.ateapi.v1.RevertActorResponse.actor:type_name -> substrate.ateapi.v1.Actor + 35, // 101: substrate.ateapi.v1.DeleteActorRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 93, // 102: substrate.ateapi.v1.DeleteActorRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions + 35, // 103: substrate.ateapi.v1.GetActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 104: substrate.ateapi.v1.CreateActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 18, // 105: substrate.ateapi.v1.CreateActorEgressPolicyRequest.egress_policy:type_name -> substrate.ateapi.v1.EgressPolicy + 35, // 106: substrate.ateapi.v1.UpdateActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 18, // 107: substrate.ateapi.v1.UpdateActorEgressPolicyRequest.egress_policy:type_name -> substrate.ateapi.v1.EgressPolicy + 35, // 108: substrate.ateapi.v1.DeleteActorEgressPolicyRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 93, // 109: substrate.ateapi.v1.DeleteActorEgressPolicyRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions + 35, // 110: substrate.ateapi.v1.GetTagRequest.tag:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 111: substrate.ateapi.v1.MintActorJwtRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 154, // 112: substrate.ateapi.v1.MintActorJwtResponse.expire_time:type_name -> google.protobuf.Timestamp + 94, // 113: substrate.ateapi.v1.ListTagsRequest.options:type_name -> substrate.ateapi.v1.ListOptions + 33, // 114: substrate.ateapi.v1.ListTagsResponse.tags:type_name -> substrate.ateapi.v1.Tag + 33, // 115: substrate.ateapi.v1.CreateTagRequest.tag:type_name -> substrate.ateapi.v1.Tag + 33, // 116: substrate.ateapi.v1.UpdateTagRequest.tag:type_name -> substrate.ateapi.v1.Tag + 35, // 117: substrate.ateapi.v1.DeleteTagRequest.tag:type_name -> substrate.ateapi.v1.ObjectRef + 93, // 118: substrate.ateapi.v1.DeleteTagRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions + 35, // 119: substrate.ateapi.v1.ListWorkerActorAssignmentsRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 94, // 120: substrate.ateapi.v1.ListWorkerActorAssignmentsRequest.options:type_name -> substrate.ateapi.v1.ListOptions + 120, // 121: substrate.ateapi.v1.ListWorkerActorAssignmentsResponse.actor_assignments:type_name -> substrate.ateapi.v1.ActorAssignment + 94, // 122: substrate.ateapi.v1.ListWorkersRequest.options:type_name -> substrate.ateapi.v1.ListOptions + 111, // 123: substrate.ateapi.v1.ListWorkersResponse.workers:type_name -> substrate.ateapi.v1.Worker + 35, // 124: substrate.ateapi.v1.GetWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 111, // 125: substrate.ateapi.v1.CreateWorkerRequest.worker:type_name -> substrate.ateapi.v1.Worker + 111, // 126: substrate.ateapi.v1.UpdateWorkerRequest.worker:type_name -> substrate.ateapi.v1.Worker + 35, // 127: substrate.ateapi.v1.DeleteWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 93, // 128: substrate.ateapi.v1.DeleteWorkerRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions + 35, // 129: substrate.ateapi.v1.DrainWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 111, // 130: substrate.ateapi.v1.DrainWorkerResponse.worker:type_name -> substrate.ateapi.v1.Worker + 35, // 131: substrate.ateapi.v1.MarkWorkerAvailableRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 111, // 132: substrate.ateapi.v1.MarkWorkerAvailableResponse.worker:type_name -> substrate.ateapi.v1.Worker + 35, // 133: substrate.ateapi.v1.MarkWorkerUnavailableRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 111, // 134: substrate.ateapi.v1.MarkWorkerUnavailableResponse.worker:type_name -> substrate.ateapi.v1.Worker + 94, // 135: substrate.ateapi.v1.ListActorsRequest.options:type_name -> substrate.ateapi.v1.ListOptions + 17, // 136: substrate.ateapi.v1.ListActorsResponse.actors:type_name -> substrate.ateapi.v1.Actor + 14, // 137: substrate.ateapi.v1.Worker.metadata:type_name -> substrate.ateapi.v1.ResourceMetadata + 153, // 138: substrate.ateapi.v1.Worker.labels:type_name -> substrate.ateapi.v1.Worker.LabelsEntry + 112, // 139: substrate.ateapi.v1.Worker.status:type_name -> substrate.ateapi.v1.WorkerStatus + 8, // 140: substrate.ateapi.v1.WorkerStatus.state:type_name -> substrate.ateapi.v1.WorkerState + 119, // 141: substrate.ateapi.v1.WorkerStatus.capacity:type_name -> substrate.ateapi.v1.WorkerResources + 119, // 142: substrate.ateapi.v1.WorkerStatus.allocated:type_name -> substrate.ateapi.v1.WorkerResources + 113, // 143: substrate.ateapi.v1.WorkerStatus.sandbox_classes:type_name -> substrate.ateapi.v1.WorkerSandboxClass + 114, // 144: substrate.ateapi.v1.WorkerSandboxClass.default_runtime:type_name -> substrate.ateapi.v1.WorkerSandboxRuntime + 114, // 145: substrate.ateapi.v1.WorkerSandboxClass.restorable_runtimes:type_name -> substrate.ateapi.v1.WorkerSandboxRuntime + 117, // 146: substrate.ateapi.v1.WorkerSandboxRuntime.version:type_name -> substrate.ateapi.v1.VersionedSandboxCompat + 115, // 147: substrate.ateapi.v1.WorkerSandboxRuntime.actor_overhead:type_name -> substrate.ateapi.v1.RuntimeOverhead + 117, // 148: substrate.ateapi.v1.ActorSandboxRuntime.version:type_name -> substrate.ateapi.v1.VersionedSandboxCompat + 118, // 149: substrate.ateapi.v1.VersionedSandboxCompat.attributes:type_name -> substrate.ateapi.v1.AttributeEntry + 14, // 150: substrate.ateapi.v1.ActorAssignment.metadata:type_name -> substrate.ateapi.v1.ResourceMetadata + 35, // 151: substrate.ateapi.v1.ActorAssignment.actor:type_name -> substrate.ateapi.v1.ObjectRef + 38, // 152: substrate.ateapi.v1.ActorAssignment.resources:type_name -> substrate.ateapi.v1.Resources + 35, // 153: substrate.ateapi.v1.AcquireWorkerEpochRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 154: substrate.ateapi.v1.RegisterWorkerRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 119, // 155: substrate.ateapi.v1.RegisterWorkerRequest.capacity:type_name -> substrate.ateapi.v1.WorkerResources + 111, // 156: substrate.ateapi.v1.RegisterWorkerResponse.worker:type_name -> substrate.ateapi.v1.Worker + 35, // 157: substrate.ateapi.v1.SetWorkerSandboxRuntimesRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 113, // 158: substrate.ateapi.v1.SetWorkerSandboxRuntimesRequest.sandbox_classes:type_name -> substrate.ateapi.v1.WorkerSandboxClass + 111, // 159: substrate.ateapi.v1.SetWorkerSandboxRuntimesResponse.worker:type_name -> substrate.ateapi.v1.Worker + 35, // 160: substrate.ateapi.v1.MintWorkerActorCertificateRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 161: substrate.ateapi.v1.MintWorkerActorCertificateRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 162: substrate.ateapi.v1.MintWorkerActorJwtRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 154, // 163: substrate.ateapi.v1.MintWorkerActorJwtResponse.expire_time:type_name -> google.protobuf.Timestamp + 35, // 164: substrate.ateapi.v1.RequestActorSuspendRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 165: substrate.ateapi.v1.RequestActorSuspendRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 17, // 166: substrate.ateapi.v1.RequestActorSuspendResponse.actor:type_name -> substrate.ateapi.v1.Actor + 35, // 167: substrate.ateapi.v1.ReportActorCrashedRequest.worker:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 168: substrate.ateapi.v1.ReportActorCrashedRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 17, // 169: substrate.ateapi.v1.ReportActorCrashedResponse.actor:type_name -> substrate.ateapi.v1.Actor + 35, // 170: substrate.ateapi.v1.ReportActorSnapshotStateRequest.actor:type_name -> substrate.ateapi.v1.ObjectRef + 137, // 171: substrate.ateapi.v1.ReportActorSnapshotStateRequest.state:type_name -> substrate.ateapi.v1.ActorSnapshotState + 138, // 172: substrate.ateapi.v1.ActorSnapshotState.latest_local:type_name -> substrate.ateapi.v1.SnapshotReport + 138, // 173: substrate.ateapi.v1.ActorSnapshotState.uploading:type_name -> substrate.ateapi.v1.SnapshotReport + 138, // 174: substrate.ateapi.v1.ActorSnapshotState.latest_durable:type_name -> substrate.ateapi.v1.SnapshotReport + 154, // 175: substrate.ateapi.v1.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp + 2, // 176: substrate.ateapi.v1.SnapshotReport.fidelity:type_name -> substrate.ateapi.v1.SnapshotFidelity + 10, // 177: substrate.ateapi.v1.SnapshotReport.object:type_name -> substrate.ateapi.v1.ObjectSnapshot + 154, // 178: substrate.ateapi.v1.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp + 154, // 179: substrate.ateapi.v1.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp + 14, // 180: substrate.ateapi.v1.AccessPolicy.metadata:type_name -> substrate.ateapi.v1.ResourceMetadata + 140, // 181: substrate.ateapi.v1.AccessPolicy.bindings:type_name -> substrate.ateapi.v1.Binding + 139, // 182: substrate.ateapi.v1.CreateGlobalAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy + 139, // 183: substrate.ateapi.v1.UpdateGlobalAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy + 35, // 184: substrate.ateapi.v1.GetAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef + 35, // 185: substrate.ateapi.v1.CreateAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef + 139, // 186: substrate.ateapi.v1.CreateAtespaceAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy + 35, // 187: substrate.ateapi.v1.UpdateAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef + 139, // 188: substrate.ateapi.v1.UpdateAtespaceAccessPolicyRequest.access_policy:type_name -> substrate.ateapi.v1.AccessPolicy + 35, // 189: substrate.ateapi.v1.DeleteAtespaceAccessPolicyRequest.atespace:type_name -> substrate.ateapi.v1.ObjectRef + 93, // 190: substrate.ateapi.v1.DeleteAtespaceAccessPolicyRequest.options:type_name -> substrate.ateapi.v1.DeleteOptions + 9, // 191: substrate.ateapi.v1.ResourceAnnotation.scope:type_name -> substrate.ateapi.v1.Scope + 155, // 192: substrate.ateapi.v1.resource:extendee -> google.protobuf.MessageOptions + 156, // 193: substrate.ateapi.v1.method:extendee -> google.protobuf.MethodOptions + 148, // 194: substrate.ateapi.v1.resource:type_name -> substrate.ateapi.v1.ResourceAnnotation + 149, // 195: substrate.ateapi.v1.method:type_name -> substrate.ateapi.v1.MethodAnnotation + 70, // 196: substrate.ateapi.v1.Control.GetActor:input_type -> substrate.ateapi.v1.GetActorRequest + 71, // 197: substrate.ateapi.v1.Control.CreateActor:input_type -> substrate.ateapi.v1.CreateActorRequest + 72, // 198: substrate.ateapi.v1.Control.UpdateActor:input_type -> substrate.ateapi.v1.UpdateActorRequest + 73, // 199: substrate.ateapi.v1.Control.SuspendActor:input_type -> substrate.ateapi.v1.SuspendActorRequest + 76, // 200: substrate.ateapi.v1.Control.ResumeActor:input_type -> substrate.ateapi.v1.ResumeActorRequest + 78, // 201: substrate.ateapi.v1.Control.RevertActor:input_type -> substrate.ateapi.v1.RevertActorRequest + 80, // 202: substrate.ateapi.v1.Control.DeleteActor:input_type -> substrate.ateapi.v1.DeleteActorRequest + 81, // 203: substrate.ateapi.v1.Control.GetActorEgressPolicy:input_type -> substrate.ateapi.v1.GetActorEgressPolicyRequest + 82, // 204: substrate.ateapi.v1.Control.CreateActorEgressPolicy:input_type -> substrate.ateapi.v1.CreateActorEgressPolicyRequest + 83, // 205: substrate.ateapi.v1.Control.UpdateActorEgressPolicy:input_type -> substrate.ateapi.v1.UpdateActorEgressPolicyRequest + 84, // 206: substrate.ateapi.v1.Control.DeleteActorEgressPolicy:input_type -> substrate.ateapi.v1.DeleteActorEgressPolicyRequest + 86, // 207: substrate.ateapi.v1.Control.MintActorJwt:input_type -> substrate.ateapi.v1.MintActorJwtRequest + 90, // 208: substrate.ateapi.v1.Control.CreateTag:input_type -> substrate.ateapi.v1.CreateTagRequest + 85, // 209: substrate.ateapi.v1.Control.GetTag:input_type -> substrate.ateapi.v1.GetTagRequest + 88, // 210: substrate.ateapi.v1.Control.ListTags:input_type -> substrate.ateapi.v1.ListTagsRequest + 91, // 211: substrate.ateapi.v1.Control.UpdateTag:input_type -> substrate.ateapi.v1.UpdateTagRequest + 92, // 212: substrate.ateapi.v1.Control.DeleteTag:input_type -> substrate.ateapi.v1.DeleteTagRequest + 97, // 213: substrate.ateapi.v1.Control.ListWorkers:input_type -> substrate.ateapi.v1.ListWorkersRequest + 99, // 214: substrate.ateapi.v1.Control.GetWorker:input_type -> substrate.ateapi.v1.GetWorkerRequest + 100, // 215: substrate.ateapi.v1.Control.CreateWorker:input_type -> substrate.ateapi.v1.CreateWorkerRequest + 101, // 216: substrate.ateapi.v1.Control.UpdateWorker:input_type -> substrate.ateapi.v1.UpdateWorkerRequest + 102, // 217: substrate.ateapi.v1.Control.DeleteWorker:input_type -> substrate.ateapi.v1.DeleteWorkerRequest + 103, // 218: substrate.ateapi.v1.Control.DrainWorker:input_type -> substrate.ateapi.v1.DrainWorkerRequest + 105, // 219: substrate.ateapi.v1.Control.MarkWorkerAvailable:input_type -> substrate.ateapi.v1.MarkWorkerAvailableRequest + 107, // 220: substrate.ateapi.v1.Control.MarkWorkerUnavailable:input_type -> substrate.ateapi.v1.MarkWorkerUnavailableRequest + 95, // 221: substrate.ateapi.v1.Control.ListWorkerActorAssignments:input_type -> substrate.ateapi.v1.ListWorkerActorAssignmentsRequest + 109, // 222: substrate.ateapi.v1.Control.ListActors:input_type -> substrate.ateapi.v1.ListActorsRequest + 60, // 223: substrate.ateapi.v1.Control.CreateAtespace:input_type -> substrate.ateapi.v1.CreateAtespaceRequest + 61, // 224: substrate.ateapi.v1.Control.GetAtespace:input_type -> substrate.ateapi.v1.GetAtespaceRequest + 62, // 225: substrate.ateapi.v1.Control.ListAtespaces:input_type -> substrate.ateapi.v1.ListAtespacesRequest + 64, // 226: substrate.ateapi.v1.Control.DeleteAtespace:input_type -> substrate.ateapi.v1.DeleteAtespaceRequest + 65, // 227: substrate.ateapi.v1.Control.CreateActorTemplate:input_type -> substrate.ateapi.v1.CreateActorTemplateRequest + 66, // 228: substrate.ateapi.v1.Control.GetActorTemplate:input_type -> substrate.ateapi.v1.GetActorTemplateRequest + 67, // 229: substrate.ateapi.v1.Control.ListActorTemplates:input_type -> substrate.ateapi.v1.ListActorTemplatesRequest + 69, // 230: substrate.ateapi.v1.Control.DeleteActorTemplate:input_type -> substrate.ateapi.v1.DeleteActorTemplateRequest + 141, // 231: substrate.ateapi.v1.Control.GetGlobalAccessPolicy:input_type -> substrate.ateapi.v1.GetGlobalAccessPolicyRequest + 142, // 232: substrate.ateapi.v1.Control.CreateGlobalAccessPolicy:input_type -> substrate.ateapi.v1.CreateGlobalAccessPolicyRequest + 143, // 233: substrate.ateapi.v1.Control.UpdateGlobalAccessPolicy:input_type -> substrate.ateapi.v1.UpdateGlobalAccessPolicyRequest + 144, // 234: substrate.ateapi.v1.Control.GetAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.GetAtespaceAccessPolicyRequest + 145, // 235: substrate.ateapi.v1.Control.CreateAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.CreateAtespaceAccessPolicyRequest + 146, // 236: substrate.ateapi.v1.Control.UpdateAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.UpdateAtespaceAccessPolicyRequest + 147, // 237: substrate.ateapi.v1.Control.DeleteAtespaceAccessPolicy:input_type -> substrate.ateapi.v1.DeleteAtespaceAccessPolicyRequest + 121, // 238: substrate.ateapi.v1.WorkerService.AcquireWorkerEpoch:input_type -> substrate.ateapi.v1.AcquireWorkerEpochRequest + 123, // 239: substrate.ateapi.v1.WorkerService.RegisterWorker:input_type -> substrate.ateapi.v1.RegisterWorkerRequest + 125, // 240: substrate.ateapi.v1.WorkerService.SetWorkerSandboxRuntimes:input_type -> substrate.ateapi.v1.SetWorkerSandboxRuntimesRequest + 127, // 241: substrate.ateapi.v1.WorkerService.MintWorkerActorCertificate:input_type -> substrate.ateapi.v1.MintWorkerActorCertificateRequest + 129, // 242: substrate.ateapi.v1.WorkerService.MintWorkerActorJwt:input_type -> substrate.ateapi.v1.MintWorkerActorJwtRequest + 131, // 243: substrate.ateapi.v1.WorkerService.RequestActorSuspend:input_type -> substrate.ateapi.v1.RequestActorSuspendRequest + 135, // 244: substrate.ateapi.v1.WorkerService.ReportActorSnapshotState:input_type -> substrate.ateapi.v1.ReportActorSnapshotStateRequest + 133, // 245: substrate.ateapi.v1.WorkerService.ReportActorCrashed:input_type -> substrate.ateapi.v1.ReportActorCrashedRequest + 17, // 246: substrate.ateapi.v1.Control.GetActor:output_type -> substrate.ateapi.v1.Actor + 17, // 247: substrate.ateapi.v1.Control.CreateActor:output_type -> substrate.ateapi.v1.Actor + 17, // 248: substrate.ateapi.v1.Control.UpdateActor:output_type -> substrate.ateapi.v1.Actor + 74, // 249: substrate.ateapi.v1.Control.SuspendActor:output_type -> substrate.ateapi.v1.SuspendActorResponse + 77, // 250: substrate.ateapi.v1.Control.ResumeActor:output_type -> substrate.ateapi.v1.ResumeActorResponse + 79, // 251: substrate.ateapi.v1.Control.RevertActor:output_type -> substrate.ateapi.v1.RevertActorResponse + 17, // 252: substrate.ateapi.v1.Control.DeleteActor:output_type -> substrate.ateapi.v1.Actor + 18, // 253: substrate.ateapi.v1.Control.GetActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy + 18, // 254: substrate.ateapi.v1.Control.CreateActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy + 18, // 255: substrate.ateapi.v1.Control.UpdateActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy + 18, // 256: substrate.ateapi.v1.Control.DeleteActorEgressPolicy:output_type -> substrate.ateapi.v1.EgressPolicy + 87, // 257: substrate.ateapi.v1.Control.MintActorJwt:output_type -> substrate.ateapi.v1.MintActorJwtResponse + 33, // 258: substrate.ateapi.v1.Control.CreateTag:output_type -> substrate.ateapi.v1.Tag + 33, // 259: substrate.ateapi.v1.Control.GetTag:output_type -> substrate.ateapi.v1.Tag + 89, // 260: substrate.ateapi.v1.Control.ListTags:output_type -> substrate.ateapi.v1.ListTagsResponse + 33, // 261: substrate.ateapi.v1.Control.UpdateTag:output_type -> substrate.ateapi.v1.Tag + 33, // 262: substrate.ateapi.v1.Control.DeleteTag:output_type -> substrate.ateapi.v1.Tag + 98, // 263: substrate.ateapi.v1.Control.ListWorkers:output_type -> substrate.ateapi.v1.ListWorkersResponse + 111, // 264: substrate.ateapi.v1.Control.GetWorker:output_type -> substrate.ateapi.v1.Worker + 111, // 265: substrate.ateapi.v1.Control.CreateWorker:output_type -> substrate.ateapi.v1.Worker + 111, // 266: substrate.ateapi.v1.Control.UpdateWorker:output_type -> substrate.ateapi.v1.Worker + 111, // 267: substrate.ateapi.v1.Control.DeleteWorker:output_type -> substrate.ateapi.v1.Worker + 104, // 268: substrate.ateapi.v1.Control.DrainWorker:output_type -> substrate.ateapi.v1.DrainWorkerResponse + 106, // 269: substrate.ateapi.v1.Control.MarkWorkerAvailable:output_type -> substrate.ateapi.v1.MarkWorkerAvailableResponse + 108, // 270: substrate.ateapi.v1.Control.MarkWorkerUnavailable:output_type -> substrate.ateapi.v1.MarkWorkerUnavailableResponse + 96, // 271: substrate.ateapi.v1.Control.ListWorkerActorAssignments:output_type -> substrate.ateapi.v1.ListWorkerActorAssignmentsResponse + 110, // 272: substrate.ateapi.v1.Control.ListActors:output_type -> substrate.ateapi.v1.ListActorsResponse + 34, // 273: substrate.ateapi.v1.Control.CreateAtespace:output_type -> substrate.ateapi.v1.Atespace + 34, // 274: substrate.ateapi.v1.Control.GetAtespace:output_type -> substrate.ateapi.v1.Atespace + 63, // 275: substrate.ateapi.v1.Control.ListAtespaces:output_type -> substrate.ateapi.v1.ListAtespacesResponse + 34, // 276: substrate.ateapi.v1.Control.DeleteAtespace:output_type -> substrate.ateapi.v1.Atespace + 36, // 277: substrate.ateapi.v1.Control.CreateActorTemplate:output_type -> substrate.ateapi.v1.ActorTemplate + 36, // 278: substrate.ateapi.v1.Control.GetActorTemplate:output_type -> substrate.ateapi.v1.ActorTemplate + 68, // 279: substrate.ateapi.v1.Control.ListActorTemplates:output_type -> substrate.ateapi.v1.ListActorTemplatesResponse + 36, // 280: substrate.ateapi.v1.Control.DeleteActorTemplate:output_type -> substrate.ateapi.v1.ActorTemplate + 139, // 281: substrate.ateapi.v1.Control.GetGlobalAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy + 139, // 282: substrate.ateapi.v1.Control.CreateGlobalAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy + 139, // 283: substrate.ateapi.v1.Control.UpdateGlobalAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy + 139, // 284: substrate.ateapi.v1.Control.GetAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy + 139, // 285: substrate.ateapi.v1.Control.CreateAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy + 139, // 286: substrate.ateapi.v1.Control.UpdateAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy + 139, // 287: substrate.ateapi.v1.Control.DeleteAtespaceAccessPolicy:output_type -> substrate.ateapi.v1.AccessPolicy + 122, // 288: substrate.ateapi.v1.WorkerService.AcquireWorkerEpoch:output_type -> substrate.ateapi.v1.AcquireWorkerEpochResponse + 124, // 289: substrate.ateapi.v1.WorkerService.RegisterWorker:output_type -> substrate.ateapi.v1.RegisterWorkerResponse + 126, // 290: substrate.ateapi.v1.WorkerService.SetWorkerSandboxRuntimes:output_type -> substrate.ateapi.v1.SetWorkerSandboxRuntimesResponse + 128, // 291: substrate.ateapi.v1.WorkerService.MintWorkerActorCertificate:output_type -> substrate.ateapi.v1.MintWorkerActorCertificateResponse + 130, // 292: substrate.ateapi.v1.WorkerService.MintWorkerActorJwt:output_type -> substrate.ateapi.v1.MintWorkerActorJwtResponse + 132, // 293: substrate.ateapi.v1.WorkerService.RequestActorSuspend:output_type -> substrate.ateapi.v1.RequestActorSuspendResponse + 136, // 294: substrate.ateapi.v1.WorkerService.ReportActorSnapshotState:output_type -> substrate.ateapi.v1.ReportActorSnapshotStateResponse + 134, // 295: substrate.ateapi.v1.WorkerService.ReportActorCrashed:output_type -> substrate.ateapi.v1.ReportActorCrashedResponse + 246, // [246:296] is the sub-list for method output_type + 196, // [196:246] is the sub-list for method input_type + 194, // [194:196] is the sub-list for extension type_name + 192, // [192:194] is the sub-list for extension extendee + 0, // [0:192] is the sub-list for field type_name } func init() { file_api_ateapi_v1_ateapi_proto_init() } @@ -11136,7 +11239,7 @@ func file_api_ateapi_v1_ateapi_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_api_ateapi_v1_ateapi_proto_rawDesc), len(file_api_ateapi_v1_ateapi_proto_rawDesc)), NumEnums: 10, - NumMessages: 143, + NumMessages: 144, NumExtensions: 2, NumServices: 2, }, diff --git a/api/ateapi/v1/ateapi.proto b/api/ateapi/v1/ateapi.proto index 51ed7d1e32..2e8a28e721 100644 --- a/api/ateapi/v1/ateapi.proto +++ b/api/ateapi/v1/ateapi.proto @@ -1845,6 +1845,24 @@ message ResumeActorRequest { // +k8s:required // +k8s:subfield(atespace)=+k8s:required ObjectRef actor = 1; + + // Resume only if this is still the actor and checkpoint the caller observed. + // This guards the current restore source; it does not select an older one. + // Restoring requires SUSPENDED or retryable RESUMING and a matching completed + // durable checkpoint, even when restoring a local copy. If its node was + // deleted, resume attempts that checkpoint's upload even if completion was + // not reported. A missing upload moves the actor to CRASHED and returns + // FAILED_PRECONDITION; call RevertActor before resuming again. + // A RUNNING actor with a matching UID and newest held checkpoint returns + // resumed=false. + // A UID or checkpoint mismatch returns FAILED_PRECONDITION: the caller's + // view is stale, so retrying the same request won't help. An invalid state + // also returns FAILED_PRECONDITION. + // A concurrent template change returns ABORTED and can be retried; a matching + // pending upload on a node not known to be deleted returns UNAVAILABLE. + // Omit for unconditional resume. + // +k8s:optional + ExpectedSnapshot expected = 2; } message ResumeActorResponse { @@ -3266,3 +3284,21 @@ extend google.protobuf.MessageOptions { extend google.protobuf.MethodOptions { MethodAnnotation method = 50000; } + +// ExpectedSnapshot identifies the actor incarnation and checkpoint to resume. +message ExpectedSnapshot { + // UID of the actor the caller observed. + // +k8s:required + // +k8s:format=k8s-uuid + string actor_uid = 1; + + // UUID of the checkpoint the caller observed. + // +k8s:required + // +k8s:format=k8s-short-name + string snapshot_uuid = 2; + + // Generation of the checkpoint the caller observed. + // +k8s:required + // +k8s:minimum=1 + int32 generation = 3; +} diff --git a/cmd/ateapi/internal/apivalidation/actor_test.go b/cmd/ateapi/internal/apivalidation/actor_test.go index 3c5595f15c..0200e52331 100644 --- a/cmd/ateapi/internal/apivalidation/actor_test.go +++ b/cmd/ateapi/internal/apivalidation/actor_test.go @@ -924,11 +924,41 @@ func TestValidateDeleteActorRequest(t *testing.T) { } func TestValidateResumeActorRequest(t *testing.T) { + withExpected := func(expected *ateapi.ExpectedSnapshot) *ateapi.ResumeActorRequest { + return &ateapi.ResumeActorRequest{ + Actor: &ateapi.ObjectRef{Atespace: "ns1", Name: "id1"}, + Expected: expected, + } + } tests := []struct { name string req *ateapi.ResumeActorRequest want field.ErrorList }{{ + "valid expected", + withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "snap-1", Generation: 1}), + nil, + }, { + "empty expected", + withExpected(&ateapi.ExpectedSnapshot{}), + field.ErrorList{ + field.Required(field.NewPath("expected", "actor_uid"), ""), + field.Required(field.NewPath("expected", "snapshot_uuid"), ""), + field.Required(field.NewPath("expected", "generation"), ""), + }, + }, { + "expected generation zero", + withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "snap-1", Generation: 0}), + field.ErrorList{field.Required(field.NewPath("expected", "generation"), "")}, + }, { + "negative expected generation", + withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "snap-1", Generation: -1}), + field.ErrorList{field.Invalid(field.NewPath("expected", "generation"), nil, "").WithOrigin("minimum")}, + }, { + "malformed expected snapshot_uuid", + withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "Bad/Snapshot", Generation: 1}), + field.ErrorList{field.Invalid(field.NewPath("expected", "snapshot_uuid"), "Bad/Snapshot", "").WithOrigin("format=k8s-short-name")}, + }, { "valid", &ateapi.ResumeActorRequest{Actor: &ateapi.ObjectRef{Atespace: "ns1", Name: "id1"}}, nil, diff --git a/cmd/ateapi/internal/apivalidation/zz_generated.validation.go b/cmd/ateapi/internal/apivalidation/zz_generated.validation.go index 4719d8d4f4..c0df34d268 100644 --- a/cmd/ateapi/internal/apivalidation/zz_generated.validation.go +++ b/cmd/ateapi/internal/apivalidation/zz_generated.validation.go @@ -3898,6 +3898,111 @@ func Validate_EnvVar( return errs } +// Validate_ExpectedSnapshot validates an instance of ExpectedSnapshot according +// to declarative validation rules in the API schema. +func Validate_ExpectedSnapshot( + ctx context.Context, op operation.Operation, fldPath *field.Path, + obj, oldObj *v1.ExpectedSnapshot) (errs field.ErrorList) { + + { // field v1.ExpectedSnapshot.ActorUid + fn := func( + fldPath *field.Path, + obj, oldObj *string, + oldValueCorrelated bool) (errs field.ErrorList) { + // don't revalidate unchanged data + if oldValueCorrelated && op.Type == operation.Update { + if obj == oldObj || (obj != nil && oldObj != nil && *obj == *oldObj) { + return nil + } + } + // call field-attached validations + earlyReturn := false + if e := validate.RequiredValue(ctx, op, fldPath, obj, oldObj).MarkShortCircuit(); len(e) != 0 { + errs = append(errs, e...) + earlyReturn = true + } + if earlyReturn { + return // do not proceed + } + if e := validate.UUID(ctx, op, fldPath, obj, oldObj); len(e) != 0 { + errs = append(errs, e...) + } + return + } + oldVal := safe.Field(oldObj, + func(oldObj *v1.ExpectedSnapshot) *string { + return &oldObj.ActorUid + }) + errs = append(errs, fn(fldPath.Child("actor_uid"), &obj.ActorUid, oldVal, oldObj != nil)...) + } + + { // field v1.ExpectedSnapshot.SnapshotUuid + fn := func( + fldPath *field.Path, + obj, oldObj *string, + oldValueCorrelated bool) (errs field.ErrorList) { + // don't revalidate unchanged data + if oldValueCorrelated && op.Type == operation.Update { + if obj == oldObj || (obj != nil && oldObj != nil && *obj == *oldObj) { + return nil + } + } + // call field-attached validations + earlyReturn := false + if e := validate.RequiredValue(ctx, op, fldPath, obj, oldObj).MarkShortCircuit(); len(e) != 0 { + errs = append(errs, e...) + earlyReturn = true + } + if earlyReturn { + return // do not proceed + } + if e := validate.ShortName(ctx, op, fldPath, obj, oldObj); len(e) != 0 { + errs = append(errs, e...) + } + return + } + oldVal := safe.Field(oldObj, + func(oldObj *v1.ExpectedSnapshot) *string { + return &oldObj.SnapshotUuid + }) + errs = append(errs, fn(fldPath.Child("snapshot_uuid"), &obj.SnapshotUuid, oldVal, oldObj != nil)...) + } + + { // field v1.ExpectedSnapshot.Generation + fn := func( + fldPath *field.Path, + obj, oldObj *int32, + oldValueCorrelated bool) (errs field.ErrorList) { + // don't revalidate unchanged data + if oldValueCorrelated && op.Type == operation.Update { + if obj == oldObj || (obj != nil && oldObj != nil && *obj == *oldObj) { + return nil + } + } + // call field-attached validations + earlyReturn := false + if e := validate.RequiredValue(ctx, op, fldPath, obj, oldObj).MarkShortCircuit(); len(e) != 0 { + errs = append(errs, e...) + earlyReturn = true + } + if earlyReturn { + return // do not proceed + } + if e := validate.Minimum(ctx, op, fldPath, obj, oldObj, 1); len(e) != 0 { + errs = append(errs, e...) + } + return + } + oldVal := safe.Field(oldObj, + func(oldObj *v1.ExpectedSnapshot) *int32 { + return &oldObj.Generation + }) + errs = append(errs, fn(fldPath.Child("generation"), &obj.Generation, oldVal, oldObj != nil)...) + } + + return errs +} + var unionMembershipFor_go_ate_dev_substrate_api_ateapi_v1_ExternalVolumeSource_ = validate.NewUnionMembership(validate.NewUnionMember("template")) // Validate_ExternalVolumeSource validates an instance of ExternalVolumeSource according @@ -7520,6 +7625,36 @@ func Validate_ResumeActorRequest( errs = append(errs, fn(fldPath.Child("actor"), obj.Actor, oldVal, oldObj != nil)...) } + { // field v1.ResumeActorRequest.Expected + fn := func( + fldPath *field.Path, + obj, oldObj *v1.ExpectedSnapshot, + oldValueCorrelated bool) (errs field.ErrorList) { + // don't revalidate unchanged data + if oldValueCorrelated && op.Type == operation.Update { + if ateDeepEqual(obj, oldObj) { + return nil + } + } + // call field-attached validations + earlyReturn := false + if e := validate.OptionalPointer(ctx, op, fldPath, obj, oldObj).MarkShortCircuit(); len(e) != 0 { + earlyReturn = true + } + if earlyReturn { + return // do not proceed + } + // call the type's validation function + errs = append(errs, Validate_ExpectedSnapshot(ctx, op, fldPath, obj, oldObj)...) + return + } + oldVal := safe.Field(oldObj, + func(oldObj *v1.ResumeActorRequest) *v1.ExpectedSnapshot { + return oldObj.Expected + }) + errs = append(errs, fn(fldPath.Child("expected"), obj.Expected, oldVal, oldObj != nil)...) + } + return errs } diff --git a/cmd/ateapi/internal/controlapi/actor.go b/cmd/ateapi/internal/controlapi/actor.go index 9b690a9f12..f242780964 100644 --- a/cmd/ateapi/internal/controlapi/actor.go +++ b/cmd/ateapi/internal/controlapi/actor.go @@ -488,7 +488,7 @@ func (s *RPCService) ResumeActor(ctx context.Context, req *ateapi.ResumeActorReq actorRef := resources.ActorRefFromObjectRef(req.GetActor()) setSpanActorRefAttributes(ctx, actorRef) - actor, resumed, err := s.actorWorkflow.ResumeActor(ctx, actorRef) + actor, resumed, err := s.actorWorkflow.ResumeActor(ctx, actorRef, req.GetExpected()) if err != nil { if errors.Is(err, store.ErrVersionConflict) { return nil, apierror.Aborted("concurrent update conflict, please retry") diff --git a/cmd/ateapi/internal/controlapi/actor_test.go b/cmd/ateapi/internal/controlapi/actor_test.go index 4787e0f2d5..da49996117 100644 --- a/cmd/ateapi/internal/controlapi/actor_test.go +++ b/cmd/ateapi/internal/controlapi/actor_test.go @@ -1053,7 +1053,7 @@ func TestCreateActor_GoldenTagDefault(t *testing.T) { } workflow := &ActorWorkflow{store: persistence} actorRef := resources.ActorRefFromActor(created) - loaded, _, err := workflow.loadActorForResume(ctx, actorRef) + loaded, _, err := workflow.loadActorForResume(ctx, actorRef, nil) if err != nil { t.Fatal(err) } diff --git a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go index 2d685e0e5e..77ed00712f 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go +++ b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go @@ -3258,7 +3258,7 @@ func TestResumeActor(t *testing.T) { podUID := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") name := "id1" - _, err := tc.client.CreateActor(context.Background(), &ateapi.CreateActorRequest{Actor: &ateapi.Actor{ + created, err := tc.client.CreateActor(context.Background(), &ateapi.CreateActorRequest{Actor: &ateapi.Actor{ Metadata: &ateapi.ResourceMetadata{Atespace: testAtespace, Name: name}, ActorTemplate: &ateapi.ObjectRef{Atespace: testAtespace, Name: "tmpl1"}, }}) @@ -3266,9 +3266,16 @@ func TestResumeActor(t *testing.T) { t.Fatalf("CreateActor failed: %v", err) } - _, err = tc.client.ResumeActor(context.Background(), &ateapi.ResumeActorRequest{ - Actor: &ateapi.ObjectRef{Atespace: testAtespace, Name: name}, - }) + expected := &ateapi.ExpectedSnapshot{ActorUid: created.GetMetadata().GetUid(), SnapshotUuid: goldenSnapshotName, Generation: 2} + request := &ateapi.ResumeActorRequest{Actor: &ateapi.ObjectRef{Atespace: testAtespace, Name: name}, Expected: expected} + if _, err := tc.client.ResumeActor(context.Background(), request); status.Code(err) != codes.FailedPrecondition { + t.Fatalf("ResumeActor with wrong generation = %v, want FailedPrecondition", err) + } + if tc.fakeAtelet.RestoreCalled { + t.Fatal("rejected resume called Restore") + } + expected.Generation = 1 + _, err = tc.client.ResumeActor(context.Background(), request) if err != nil { t.Fatalf("ResumeActor failed: %v", err) } diff --git a/cmd/ateapi/internal/controlapi/hosting_test.go b/cmd/ateapi/internal/controlapi/hosting_test.go index e9b2a829f7..f3558e1688 100644 --- a/cmd/ateapi/internal/controlapi/hosting_test.go +++ b/cmd/ateapi/internal/controlapi/hosting_test.go @@ -190,7 +190,7 @@ func TestHostingNodeAfterResumeOntoNewWorker(t *testing.T) { tmpl := &ateapi.ActorTemplate{ WorkerSelector: &ateapi.Selector{MatchLabels: map[string]string{"pool": "b"}}, } - resuming, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl) + resuming, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } diff --git a/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments_test.go b/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments_test.go index 802014c69e..65cf60284b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments_test.go @@ -453,7 +453,7 @@ func TestAssignWorkerAttempt_LeavesWorkerEpochUnset(t *testing.T) { w := &ActorWorkflow{store: persistence, workerCache: wc, scheduler: scheduling.New(wc)} tmpl := &ateapi.ActorTemplate{} - stored, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl) + stored, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index 324a16eda6..506dad1e11 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -51,7 +51,7 @@ type restoreTelemetry struct { // ResumeActor executes the workflow to resume a suspended actor. Idempotent: // a re-entered workflow fast-forwards past the steps a previous attempt // completed, deriving progress from the persisted actor alone. -func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapi.Actor, resumed bool, err error) { +func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.ActorRef, expected *ateapi.ExpectedSnapshot) (_ *ateapi.Actor, resumed bool, err error) { ctx = preview.ContextWithOptIn(ctx) start := time.Now() var actor *ateapi.Actor @@ -79,7 +79,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto if err != nil { return nil, false, err } - if wasRunning = actor.GetStatus().GetState() == ateapi.ActorState_ACTOR_STATE_RUNNING; wasRunning { + if wasRunning = actor.GetStatus().GetState() == ateapi.ActorState_ACTOR_STATE_RUNNING; wasRunning && expected == nil { return actor, false, nil } @@ -89,7 +89,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto } defer lease.Close() - actor, actorTemplate, err = w.loadActorForResume(leaseCtx, actorRef) + actor, actorTemplate, err = w.loadActorForResume(leaseCtx, actorRef, expected) if err != nil { return nil, false, err } @@ -103,14 +103,14 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto actor = created var worker *ateapi.Worker var assigned *ateapi.Actor - assigned, worker, err = w.ensureWorkerAssigned(leaseCtx, actorRef, actor, actorTemplate) + assigned, worker, err = w.ensureWorkerAssigned(leaseCtx, actorRef, actor, actorTemplate, expected) if err != nil && shouldDetachFromNode(actor, err) { var detached *ateapi.Actor - if detached, err = w.detachFromFullNode(leaseCtx, actorRef, actor, actorTemplate, err); err != nil { + if detached, err = w.detachFromFullNode(leaseCtx, actorRef, actor, actorTemplate, err, expected); err != nil { return nil, false, err } actor = detached - assigned, worker, err = w.ensureWorkerAssigned(leaseCtx, actorRef, actor, actorTemplate) + assigned, worker, err = w.ensureWorkerAssigned(leaseCtx, actorRef, actor, actorTemplate, expected) } if err != nil { return nil, false, err @@ -122,6 +122,10 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto } var workerEpoch int64 if tele, workerEpoch, err = w.ensureAteletRestored(leaseCtx, actorRef, actor, actorTemplate, volumePublishContexts); err != nil { + snap := snapshotToResume(actor.GetStatus()) + if expected != nil && localElsewhere(snap, actor.GetStatus().GetAssignedNode()) && apierror.Code(err) == codes.FailedPrecondition { + err = apierror.FailedPrecondition("held snapshot %q generation %d could not satisfy expected snapshot %q generation %d: %w", snap.GetUuid(), snap.GetGeneration(), expected.GetSnapshotUuid(), expected.GetGeneration(), err) + } return nil, false, err } var running *ateapi.Actor @@ -133,7 +137,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto } // loadActorForResume fetches the current actor record and its template. -func (w *ActorWorkflow) loadActorForResume(ctx context.Context, actorRef resources.ActorRef) (_ *ateapi.Actor, _ *ateapi.ActorTemplate, err error) { +func (w *ActorWorkflow) loadActorForResume(ctx context.Context, actorRef resources.ActorRef, expected *ateapi.ExpectedSnapshot) (_ *ateapi.Actor, _ *ateapi.ActorTemplate, err error) { ctx, done := stepSpan(ctx, "LoadActorForResume") defer func() { err = done(err) }() @@ -145,6 +149,10 @@ func (w *ActorWorkflow) loadActorForResume(ctx context.Context, actorRef resourc return nil, nil, fmt.Errorf("while getting actor from DB: %w", err) } + if err := w.validateExpectedSnapshot(ctx, actor, expected); err != nil { + return nil, nil, err + } + // If the actor is already running, there is no pending restore to prepare // for. Short-circuit immediately to avoid an unnecessary template read on // the hot resume path. @@ -187,6 +195,60 @@ func snapshotSharing(owner ateapi.SnapshotOwner) ateletpb.SnapshotSharing { return ateletpb.SnapshotSharing_SNAPSHOT_SHARING_PRIVATE } +// validateExpectedSnapshot checks the held checkpoint and snapshotToResume, +// not last_assigned_generation: revert can discard generations. +// Call only with an actor read under the lifecycle lease, including refreshes +// after assignment conflicts, before doing further workload work. +func (w *ActorWorkflow) validateExpectedSnapshot(ctx context.Context, actor *ateapi.Actor, expected *ateapi.ExpectedSnapshot) error { + if expected == nil { + return nil + } + if actor.GetMetadata().GetUid() != expected.GetActorUid() { + return apierror.FailedPrecondition("actor UID %q does not match expected actor UID %q", actor.GetMetadata().GetUid(), expected.GetActorUid()) + } + switch actor.GetStatus().GetState() { + case ateapi.ActorState_ACTOR_STATE_SUSPENDED, ateapi.ActorState_ACTOR_STATE_RESUMING, ateapi.ActorState_ACTOR_STATE_RUNNING: + default: + return apierror.FailedPrecondition("actor in state %s cannot resume an expected snapshot", actor.GetStatus().GetState()) + } + latest := latestSnapshot(actor.GetStatus(), func(*ateapi.Snapshot) bool { return true }) + if latest == nil || latest.GetUuid() != expected.GetSnapshotUuid() || latest.GetGeneration() != expected.GetGeneration() { + return apierror.FailedPrecondition("actor's newest held snapshot %q generation %d does not match expected snapshot %q generation %d", latest.GetUuid(), latest.GetGeneration(), expected.GetSnapshotUuid(), expected.GetGeneration()) + } + // An identical retry after a successful resume has no restore to prepare. + if actor.GetStatus().GetState() == ateapi.ActorState_ACTOR_STATE_RUNNING { + return nil + } + switch latest.GetDurableSnapshot().GetState() { + case ateapi.SnapshotStorageState_SNAPSHOT_STORAGE_STATE_PENDING: + if node := latest.GetLocality(); node != "" { + // A released node's snapshot is restored from its upload. + if localElsewhere(latest, actor.GetStatus().GetAssignedNode()) { + break + } + gone, err := w.dialer.NodeGone(ctx, node) + if err != nil { + if ctx.Err() != nil { + return fmt.Errorf("while checking expected snapshot node %q: %w: %w", node, err, ctx.Err()) + } + return apierror.Unavailable("while checking expected snapshot node %q: %w", node, err) + } + if gone { + break + } + } + return apierror.Unavailable("expected snapshot is not yet complete in durable storage") + case ateapi.SnapshotStorageState_SNAPSHOT_STORAGE_STATE_COMPLETED: + default: + return apierror.FailedPrecondition("expected snapshot has no completed durable checkpoint") + } + if source := snapshotToResume(actor.GetStatus()); source == nil || + source.GetUuid() != expected.GetSnapshotUuid() || source.GetGeneration() != expected.GetGeneration() { + return apierror.FailedPrecondition("actor's restore source %q generation %d does not match expected snapshot %q generation %d", source.GetUuid(), source.GetGeneration(), expected.GetSnapshotUuid(), expected.GetGeneration()) + } + return nil +} + // ensureVolumesCreated provisions any actor volumes that have no storage // volume ID yet, persisting the resulting volume state (even when creation // partially failed, so progress is not lost) and returning the stored copy. @@ -234,7 +296,7 @@ func (w *ActorWorkflow) ensureVolumesCreated(ctx context.Context, actorRef resou // ensureWorkerAssigned assigns a worker and leaves the actor RESUMING. // Re-entry validates the recorded assignment before reuse. // Version conflicts re-read the actor before retrying. -func (w *ActorWorkflow) ensureWorkerAssigned(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate) (_ *ateapi.Actor, _ *ateapi.Worker, err error) { +func (w *ActorWorkflow) ensureWorkerAssigned(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate, expected *ateapi.ExpectedSnapshot) (_ *ateapi.Actor, _ *ateapi.Worker, err error) { ctx, done := stepSpan(ctx, "AssignWorker") defer func() { err = done(err) }() @@ -268,7 +330,7 @@ func (w *ActorWorkflow) ensureWorkerAssigned(ctx context.Context, actorRef resou return false, err } attemptStart := time.Now() - attemptActor, attemptWorker, attemptErr := w.assignWorkerAttempt(ctx, actorRef, actor, actorTemplate) + attemptActor, attemptWorker, attemptErr := w.assignWorkerAttempt(ctx, actorRef, actor, actorTemplate, expected) lastAttempt = time.Since(attemptStart) attempted = true if attemptErr == nil { @@ -402,7 +464,7 @@ func schedulerRecordable(err error) bool { // re-reads the actor: if the fresh copy can still be resumed the refreshed // actor is returned along with the conflict so the caller retries with clean // inputs; any other status aborts the resume. -func (w *ActorWorkflow) assignWorkerAttempt(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate) (_ *ateapi.Actor, _ *ateapi.Worker, err error) { +func (w *ActorWorkflow) assignWorkerAttempt(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate, expected *ateapi.ExpectedSnapshot) (_ *ateapi.Actor, _ *ateapi.Worker, err error) { // Checked before a worker is claimed. if err := checkSnapshotUpload(actorRef, actor); err != nil { return nil, nil, err @@ -506,6 +568,13 @@ func (w *ActorWorkflow) assignWorkerAttempt(ctx context.Context, actorRef resour slog.WarnContext(ctx, "Failed to refresh actor after assignment conflict", slog.Any("err", gerr)) return nil, nil, err } + conditionErr := w.validateExpectedSnapshot(ctx, fresh, expected) + if conditionErr == nil && expected != nil && !proto.Equal(fresh.GetActorTemplate(), actor.GetActorTemplate()) { + conditionErr = apierror.Aborted("actor template changed during conditioned resume, please retry") + } + if conditionErr != nil { + return nil, nil, conditionErr + } switch fresh.GetStatus().GetState() { case ateapi.ActorState_ACTOR_STATE_SUSPENDED: slog.InfoContext(ctx, "Retrying assignment due to actor version conflict", slog.Any("actor", actorRef)) diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_detach.go b/cmd/ateapi/internal/controlapi/workflow_resume_detach.go index 680a4d5a03..fcc303d067 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_detach.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_detach.go @@ -44,13 +44,13 @@ func shouldDetachFromNode(actor *ateapi.Actor, assignErr error) bool { // Unreachable nodes can be forgotten when the latest local snapshot is durable. // A deleted node is released even when that snapshot is not durable, so resume // tries its upload. -func (w *ActorWorkflow) detachFromFullNode(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate, assignErr error) (_ *ateapi.Actor, err error) { +func (w *ActorWorkflow) detachFromFullNode(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, actorTemplate *ateapi.ActorTemplate, assignErr error, expected *ateapi.ExpectedSnapshot) (_ *ateapi.Actor, err error) { ctx, done := stepSpan(ctx, "DetachFromFullNode") defer func() { err = done(err) }() node := actor.GetStatus().GetAssignedNode() // Assignment may have retried on a newer version than actor. - if actor, err = w.loadPinnedActor(ctx, actorRef, actor.GetMetadata().GetUid(), node); err != nil { + if actor, err = w.loadPinnedActor(ctx, actorRef, actor, expected); err != nil { return nil, err } constraints := schedulingConstraints(actor, actorTemplate) @@ -80,7 +80,7 @@ func (w *ActorWorkflow) detachFromFullNode(ctx context.Context, actorRef resourc return nil, apierror.Unavailable("while getting atelet conn for node %q: %w (and while checking the node: %w)", node, err, nodeErr) } if gone { - return w.recordGoneNode(ctx, actorRef, actor, node) + return w.recordGoneNode(ctx, actorRef, actor, node, expected) } } return nil, apierror.Unavailable("while getting atelet conn for node %q: %w", node, err) @@ -113,16 +113,22 @@ func (w *ActorWorkflow) detachFromFullNode(ctx context.Context, actorRef resourc return stored, nil } -// loadPinnedActor reads the actor, and fails with Aborted unless it is still -// the SUSPENDED incarnation uid, pinned to node. -func (w *ActorWorkflow) loadPinnedActor(ctx context.Context, actorRef resources.ActorRef, uid, node string) (*ateapi.Actor, error) { +// loadPinnedActor revalidates a conditioned resume before detach and fails +// with Aborted unless the actor is still SUSPENDED and pinned to the same node. +func (w *ActorWorkflow) loadPinnedActor(ctx context.Context, actorRef resources.ActorRef, previous *ateapi.Actor, expected *ateapi.ExpectedSnapshot) (*ateapi.Actor, error) { actor, err := w.store.GetActor(ctx, actorRef) if err != nil { return nil, fmt.Errorf("while reading Actor %s: %w", actorRef, err) } - if actor.GetMetadata().GetUid() != uid || + if err := w.validateExpectedSnapshot(ctx, actor, expected); err != nil { + return nil, err + } + if expected != nil && !proto.Equal(actor.GetActorTemplate(), previous.GetActorTemplate()) { + return nil, apierror.Aborted("actor template changed during conditioned resume, please retry") + } + if actor.GetMetadata().GetUid() != previous.GetMetadata().GetUid() || actor.GetStatus().GetState() != ateapi.ActorState_ACTOR_STATE_SUSPENDED || - actor.GetStatus().GetAssignedNode() != node { + actor.GetStatus().GetAssignedNode() != previous.GetStatus().GetAssignedNode() { return nil, apierror.Aborted("Actor %s changed during the resume, please retry", actorRef) } return actor, nil @@ -162,8 +168,8 @@ func (w *ActorWorkflow) recordDetach(ctx context.Context, actorRef resources.Act // reporting, so the snapshot keeps its locality and resume restores it from // the upload on another node. It reads the actor again first, so an upload // recorded during the node lookup detaches it instead. -func (w *ActorWorkflow) recordGoneNode(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, node string) (*ateapi.Actor, error) { - actor, err := w.loadPinnedActor(ctx, actorRef, actor.GetMetadata().GetUid(), node) +func (w *ActorWorkflow) recordGoneNode(ctx context.Context, actorRef resources.ActorRef, actor *ateapi.Actor, node string, expected *ateapi.ExpectedSnapshot) (*ateapi.Actor, error) { + actor, err := w.loadPinnedActor(ctx, actorRef, actor, expected) if err != nil { return nil, err } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_detach_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_detach_test.go index bcbdd06432..317b9e81b6 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_detach_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_detach_test.go @@ -432,7 +432,7 @@ func TestDetachFromFullNode(t *testing.T) { t.Fatalf("GetActor: %v", err) } - stored, err := w.detachFromFullNode(ctx, actorRef, actor, &ateapi.ActorTemplate{}, noRoom) + stored, err := w.detachFromFullNode(ctx, actorRef, actor, &ateapi.ActorTemplate{}, noRoom, nil) if got := apierror.Code(err); got != tt.wantCode { t.Fatalf("detachFromFullNode() = %v, want code %v", err, tt.wantCode) } @@ -492,12 +492,18 @@ func TestDetachFromFullNode(t *testing.T) { // room on node-2: it moves to node-2 only when node-1 is full. func TestResumeActor_DetachesFromFullNode(t *testing.T) { tests := []struct { - name string - node1 bool - wantNode string + name string + node1 bool + wantNode string + conditioned bool + change func(*ateapi.Actor) + want codes.Code }{ {name: "node-1 full: detaches and resumes on node-2", wantNode: "node-2"}, {name: "node-1 has room: resumes there from the local copy", node1: true, wantNode: "node-1"}, + {name: "expected checkpoint detaches and resumes", conditioned: true, wantNode: "node-2"}, + {name: "template changes before detach then identical retry succeeds", conditioned: true, change: func(a *ateapi.Actor) { a.ActorTemplate.Name = "other" }, want: codes.Aborted, wantNode: "node-2"}, + {name: "snapshot changes before detach", conditioned: true, change: func(a *ateapi.Actor) { a.Status.Snapshots[0].Uuid = "other" }, want: codes.FailedPrecondition}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -520,6 +526,9 @@ func TestResumeActor_DetachesFromFullNode(t *testing.T) { fillWorker(t, st, workers["node-1"]) } snap := newPausedSnapshot(t, 1, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", "pause-snap-1", "node-1") + if tt.conditioned { + snap.DurableSnapshot.State = ateapi.SnapshotStorageState_SNAPSHOT_STORAGE_STATE_COMPLETED + } seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapi.ActorState_ACTOR_STATE_SUSPENDED, func(a *ateapi.Actor) { a.Status.AssignedNode = "node-1" a.Status.LastAssignedGeneration = 1 @@ -529,9 +538,54 @@ func TestResumeActor_DetachesFromFullNode(t *testing.T) { addAteletOnNode(t, w, "node-2") atelet.detachState = detachedState(snap) - got, resumed, err := w.ResumeActor(ctx, actorRef) - if err != nil { - t.Fatalf("ResumeActor: %v", err) + actor := mustReadActor(t, ctx, st, actorRef) + var expected *ateapi.ExpectedSnapshot + if tt.conditioned { + expected = &ateapi.ExpectedSnapshot{ActorUid: actor.GetMetadata().GetUid(), SnapshotUuid: snap.GetUuid(), Generation: snap.GetGeneration()} + } + var changed *ateapi.Actor + if tt.change != nil { + w.store = &detachReadChangingStore{Interface: st, beforeDetachRead: func() { + var err error + changed, err = st.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(a *ateapi.Actor) error { tt.change(a); return nil }) + if err != nil { + t.Fatal(err) + } + if tt.want == codes.Aborted { + if _, err := st.CreateActorTemplate(ctx, &ateapi.ActorTemplate{ + Metadata: &ateapi.ResourceMetadata{Atespace: "ns", Name: "other"}, + SnapshotConfig: &ateapi.SnapshotConfig{StorageLocation: testStorageLocation}, + }); err != nil { + t.Fatal(err) + } + } else { + // Validate the condition even when no atelet is available. + if err := w.dialer.ateletIndexer.Replace(nil, ""); err != nil { + t.Fatal(err) + } + } + }} + } + got, resumed, err := w.ResumeActor(ctx, actorRef, expected) + if apierror.Code(err) != tt.want { + t.Fatalf("ResumeActor = %v, want %v", err, tt.want) + } + if tt.want != codes.OK { + if changed == nil { + t.Fatal("did not inject change at detach reread") + } + restore, run := atelet.requests() + if atelet.detachRequest() != nil || restore != nil || run != nil || !proto.Equal(mustReadActor(t, ctx, st, actorRef), changed) { + t.Error("rejected resume called atelet or changed actor") + } + if tt.want != codes.Aborted { + return + } + w.store = st + got, resumed, err = w.ResumeActor(ctx, actorRef, expected) + if err != nil { + t.Fatalf("identical retry after template change: %v", err) + } } if !resumed || got.GetStatus().GetState() != ateapi.ActorState_ACTOR_STATE_RUNNING { t.Fatalf("ResumeActor = (state %v, resumed %v), want RUNNING and resumed", got.GetStatus().GetState(), resumed) @@ -583,8 +637,27 @@ func TestResumeActor_NodeGoneWithPendingUpload(t *testing.T) { uploaded bool // otherRuntime gives the only free worker a runtime other than the // snapshot's, so the restore falls back to ROOTFS. - otherRuntime bool + otherRuntime bool + conditioned bool + resuming bool + nodePresent bool + registered bool + noNodeGetter bool + nodeLookupErr error + cancelLookup bool + staleExpected bool + preflight codes.Code }{ + {name: "conditioned: live node still waits for upload", assignedNode: "node-3", conditioned: true, nodePresent: true, preflight: codes.Unavailable}, + {name: "conditioned: registered atelet still waits for upload", assignedNode: "node-3", conditioned: true, registered: true, preflight: codes.Unavailable}, + {name: "conditioned: no node getter still waits for upload", assignedNode: "node-3", conditioned: true, noNodeGetter: true, preflight: codes.Unavailable}, + {name: "conditioned: node lookup failure is retryable", assignedNode: "node-3", conditioned: true, nodeLookupErr: errors.New("node lookup unavailable"), preflight: codes.Unavailable}, + {name: "conditioned: node lookup cancellation is preserved", assignedNode: "node-3", conditioned: true, cancelLookup: true, preflight: codes.Canceled}, + {name: "conditioned: stale expectation fails before node lookup", assignedNode: "node-3", conditioned: true, staleExpected: true, preflight: codes.FailedPrecondition}, + {name: "conditioned: restores the matching upload", assignedNode: "node-3", uploaded: true, conditioned: true}, + {name: "conditioned: missing upload fails permanently", assignedNode: "node-3", conditioned: true}, + {name: "conditioned: already released node restores the matching upload", uploaded: true, conditioned: true}, + {name: "conditioned: RESUMING retry restores the matching upload", uploaded: true, conditioned: true, resuming: true}, {name: "uploaded: restores on another node", assignedNode: "node-3", uploaded: true}, {name: "released by an earlier resume: restores on another node", uploaded: true}, {name: "uploaded, only another runtime has room: restores at ROOTFS", assignedNode: "node-3", uploaded: true, otherRuntime: true}, @@ -628,8 +701,78 @@ func TestResumeActor_NodeGoneWithPendingUpload(t *testing.T) { atelet.restoreErr = apierror.NotFound("snapshot has no usable local copy and no uploaded copy") } - got, resumed, err := w.ResumeActor(ctx, actorRef) + actor := mustReadActor(t, ctx, st, actorRef) + var expected *ateapi.ExpectedSnapshot + if tt.conditioned { + expected = &ateapi.ExpectedSnapshot{ActorUid: actor.GetMetadata().GetUid(), SnapshotUuid: "pending-2", Generation: 2} + } + if tt.resuming { + tmpl, err := resolveActorTemplate(ctx, st, actor) + if err != nil { + t.Fatal(err) + } + if _, _, err := w.ensureWorkerAssigned(ctx, actorRef, actor, tmpl, nil); err != nil { + t.Fatalf("persist RESUMING: %v", err) + } + // The old node name has been reused since the actor was released. + WithNodeGetter(nodeGetter(nil, "node-1", "node-2", "node-3"))(w.dialer) + } + callCtx, cancel := context.WithCancel(ctx) + defer cancel() + nodeReads := 0 + if tt.preflight != codes.OK { + getter := nodeGetter(tt.nodeLookupErr, "node-1", "node-2") + if tt.nodePresent { + getter = nodeGetter(nil, "node-1", "node-2", "node-3") + } + WithNodeGetter(func(ctx context.Context, name string) error { + nodeReads++ + if tt.cancelLookup { + cancel() + return ctx.Err() + } + return getter(ctx, name) + })(w.dialer) + if tt.noNodeGetter { + WithNodeGetter(nil)(w.dialer) + } + if tt.registered { + addAteletOnNode(t, w, "node-3") + } + if tt.staleExpected { + expected.SnapshotUuid, expected.Generation = "durable-1", 1 + } + } + got, resumed, err := w.ResumeActor(callCtx, actorRef, expected) + if tt.preflight != codes.OK { + if apierror.Code(err) != tt.preflight { + t.Fatalf("ResumeActor = %v, want %v", err, tt.preflight) + } + if tt.cancelLookup && !errors.Is(err, context.Canceled) { + t.Errorf("lookup cancellation was lost: %v", err) + } + if tt.staleExpected && nodeReads != 0 { + t.Errorf("stale expectation performed %d node lookups", nodeReads) + } + restore, run := atelet.requests() + if restore != nil || run != nil || !proto.Equal(mustReadActor(t, ctx, st, actorRef), actor) { + t.Error("rejected resume changed actor or called atelet") + } + if assignment := firstAssignment(t, st, worker.GetMetadata().GetName()); assignment != nil { + t.Errorf("rejected resume retained worker assignment: %v", assignment) + } + return + } + if tt.uploaded && err != nil { + t.Fatalf("ResumeActor: %v", err) + } + if !tt.uploaded && apierror.Code(err) != codes.FailedPrecondition { + t.Fatalf("ResumeActor = %v, want FailedPrecondition", err) + } restore, _ := atelet.requests() + if restore.GetSnapshot().GetGeneration() != 2 || restore.GetSnapshot().GetObject().GetSnapshotUri() != pending.GetDurableSnapshot().GetObject().GetSnapshotUri() { + t.Errorf("Restore did not name generation 2's upload: %v", restore.GetSnapshot()) + } if restore.GetWorkerPodUid() != worker.GetWorkerPodUid() || restore.GetSnapshot().GetSnapshotUuid() != pending.GetUuid() { t.Fatalf("Restore = (worker %q, snapshot %q), want (%q, %q)", restore.GetWorkerPodUid(), restore.GetSnapshot().GetSnapshotUuid(), worker.GetWorkerPodUid(), pending.GetUuid()) } @@ -653,8 +796,15 @@ func TestResumeActor_NodeGoneWithPendingUpload(t *testing.T) { return } - if apierror.Code(err) != codes.FailedPrecondition { - t.Fatalf("ResumeActor = %v, want FailedPrecondition", err) + if tt.conditioned { + for _, detail := range []string{"held snapshot", "pending-2", "generation 2", "expected snapshot"} { + if !strings.Contains(err.Error(), detail) { + t.Errorf("missing upload error %q omits %q", err, detail) + } + } + if _, _, retryErr := w.ResumeActor(ctx, actorRef, expected); apierror.Code(retryErr) != codes.FailedPrecondition { + t.Errorf("retry after missing upload = %v, want FailedPrecondition", retryErr) + } } crashed, err := st.GetActor(ctx, actorRef) if err != nil { @@ -682,7 +832,7 @@ func TestResumeActor_NodeGoneWithPendingUpload(t *testing.T) { atelet.mu.Lock() atelet.restoreErr = nil atelet.mu.Unlock() - if _, _, err := w.ResumeActor(ctx, actorRef); err != nil { + if _, _, err := w.ResumeActor(ctx, actorRef, nil); err != nil { t.Fatalf("second ResumeActor: %v", err) } if restore, _ := atelet.requests(); restore.GetSnapshot().GetSnapshotUuid() != durable.GetUuid() { @@ -692,6 +842,22 @@ func TestResumeActor_NodeGoneWithPendingUpload(t *testing.T) { } } +// detachReadChangingStore changes the actor on the read after preflight and +// the initial under-lease read, before the full-node detach can start. +type detachReadChangingStore struct { + store.Interface + reads int + beforeDetachRead func() +} + +func (s *detachReadChangingStore) GetActor(ctx context.Context, ref resources.ActorRef) (*ateapi.Actor, error) { + s.reads++ + if s.reads == 3 { + s.beforeDetachRead() + } + return s.Interface.GetActor(ctx, ref) +} + // addAteletOnNode routes node's atelet to the same capturing fake as node-1's. func addAteletOnNode(t *testing.T, w *ActorWorkflow, node string) { t.Helper() @@ -736,12 +902,12 @@ func TestDetachFromFullNode_ReentryRecordsAfterConflict(t *testing.T) { t.Errorf("UpdateActor: %v", err) } } - if _, err := w.detachFromFullNode(ctx, actorRef, actor, &ateapi.ActorTemplate{}, noRoom); apierror.Code(err) != codes.Aborted { + if _, err := w.detachFromFullNode(ctx, actorRef, actor, &ateapi.ActorTemplate{}, noRoom, nil); apierror.Code(err) != codes.Aborted { t.Fatalf("first detachFromFullNode() = %v, want Aborted", err) } atelet.onDetach = nil - stored, err := w.detachFromFullNode(ctx, actorRef, actor, &ateapi.ActorTemplate{}, noRoom) + stored, err := w.detachFromFullNode(ctx, actorRef, actor, &ateapi.ActorTemplate{}, noRoom, nil) if err != nil { t.Fatalf("second detachFromFullNode() = %v, want success", err) } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_expected_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_expected_test.go new file mode 100644 index 0000000000..fa23a9db8b --- /dev/null +++ b/cmd/ateapi/internal/controlapi/workflow_resume_expected_test.go @@ -0,0 +1,287 @@ +// Copyright 2026 The Agent Substrate Authors +// +// 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" + "strings" + "testing" + + "go.ate.dev/substrate/api/ateapi/v1" + "go.ate.dev/substrate/cmd/ateapi/internal/store" + "go.ate.dev/substrate/cmd/ateapi/internal/store/storetest" + "go.ate.dev/substrate/internal/apierror" + "go.ate.dev/substrate/internal/resources" + "google.golang.org/grpc/codes" + "google.golang.org/protobuf/proto" +) + +func TestResumeActor_ExpectedSnapshot(t *testing.T) { + pending := func(a *ateapi.Actor) { + a.Status.Snapshots[0].DurableSnapshot.State = ateapi.SnapshotStorageState_SNAPSHOT_STORAGE_STATE_PENDING + } + local := func(a *ateapi.Actor) { a.Status.AssignedNode, a.Status.Snapshots[0].Locality = "node-1", "node-1" } + newer := func(a *ateapi.Actor) { + a.Status.LastAssignedGeneration = 2 + a.Status.Snapshots = append(a.Status.Snapshots, newInProgressSnapshot(t, 2, ateapi.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "snap-2")) + } + changed := func(a *ateapi.Actor) { + a.Status.State = ateapi.ActorState_ACTOR_STATE_SUSPENDED + a.Status.Snapshots[0].Uuid, a.Status.Snapshots[0].Generation, a.Status.LastAssignedGeneration = "snap-2", 2, 2 + } + for _, tc := range []struct { + name, phase string + mutate func(*ateapi.Actor) + request func(*ateapi.ExpectedSnapshot) + apiUpdate bool + noop bool + diagnostic string + want codes.Code + }{ + {name: "matching checkpoint/durable"}, + {name: "invalid lifecycle state", mutate: func(a *ateapi.Actor) { a.Status.State = ateapi.ActorState_ACTOR_STATE_CRASHED }, diagnostic: "cannot resume an expected snapshot", want: codes.FailedPrecondition}, + {name: "no held snapshot", mutate: func(a *ateapi.Actor) { a.Status.Snapshots = nil }, want: codes.FailedPrecondition}, + {name: "local checkpoint/no durable copy", mutate: func(a *ateapi.Actor) { + local(a) + a.Status.Snapshots[0].DurableSnapshot = nil + }, want: codes.FailedPrecondition}, + {name: "local checkpoint/unspecified durable status", mutate: func(a *ateapi.Actor) { + local(a) + a.Status.Snapshots[0].DurableSnapshot.State = ateapi.SnapshotStorageState_SNAPSHOT_STORAGE_STATE_UNSPECIFIED + }, want: codes.FailedPrecondition}, + {name: "matching checkpoint/UUID mismatch", request: func(e *ateapi.ExpectedSnapshot) { e.SnapshotUuid = "other" }, diagnostic: "newest held snapshot", want: codes.FailedPrecondition}, + {name: "matching checkpoint/generation mismatch", request: func(e *ateapi.ExpectedSnapshot) { e.Generation = 2 }, diagnostic: "newest held snapshot", want: codes.FailedPrecondition}, + {name: "matching checkpoint/lost response identical retry", phase: "lost response"}, + {name: "matching checkpoint/retryable RESUMING", phase: "resuming"}, + {name: "matching checkpoint/newer durable overrides older local", mutate: func(a *ateapi.Actor) { + local(a) + next := proto.CloneOf(a.Status.Snapshots[0]) + next.Uuid, next.Generation, next.Locality = "snap-2", 2, "" + a.Status.Snapshots = append(a.Status.Snapshots, next) + a.Status.LastAssignedGeneration = 2 + }, request: func(e *ateapi.ExpectedSnapshot) { e.SnapshotUuid, e.Generation = "snap-2", 2 }}, + {name: "GetActor-ResumeActor race/intervening suspend", phase: "suspend", want: codes.FailedPrecondition}, + {name: "GetActor-ResumeActor race/source changes after preflight", phase: "lease", mutate: changed, want: codes.FailedPrecondition}, + {name: "same-name replacement", phase: "replace", diagnostic: "actor UID", want: codes.FailedPrecondition}, + {name: "RUNNING/matching checkpoint", mutate: func(a *ateapi.Actor) { a.Status.State = ateapi.ActorState_ACTOR_STATE_RUNNING }, noop: true}, + {name: "RUNNING/pending held checkpoint", mutate: func(a *ateapi.Actor) { + a.Status.State = ateapi.ActorState_ACTOR_STATE_RUNNING + pending(a) + }, noop: true}, + {name: "RUNNING/UUID mismatch", mutate: func(a *ateapi.Actor) { a.Status.State = ateapi.ActorState_ACTOR_STATE_RUNNING }, request: func(e *ateapi.ExpectedSnapshot) { e.SnapshotUuid = "other" }, diagnostic: "newest held snapshot", want: codes.FailedPrecondition}, + {name: "RUNNING/newer held snapshot under lease", phase: "lease", mutate: func(a *ateapi.Actor) { + a.Status.State = ateapi.ActorState_ACTOR_STATE_RUNNING + newer(a) + }, diagnostic: "newest held snapshot", want: codes.FailedPrecondition}, + {name: "newer pending checkpoint/older completed refused", mutate: newer, want: codes.FailedPrecondition}, + {name: "newer pending checkpoint/matching upload unavailable", mutate: newer, request: func(e *ateapi.ExpectedSnapshot) { e.SnapshotUuid, e.Generation = "snap-2", 2 }, want: codes.Unavailable}, + {name: "template change/assignment conflict", phase: "conflict", mutate: func(a *ateapi.Actor) { a.ActorTemplate.Name = "tmpl2" }, apiUpdate: true, want: codes.Aborted}, + {name: "matching checkpoint/assignment refresh rejects changed snapshot", phase: "conflict", mutate: func(a *ateapi.Actor) { a.Status.Snapshots[0].Uuid = "other" }, want: codes.FailedPrecondition}, + {name: "template change/unrelated update retries", phase: "conflict", mutate: func(a *ateapi.Actor) { a.WorkerSelector = &ateapi.Selector{} }}, + {name: "malformed expectation", phase: "malformed", request: func(e *ateapi.ExpectedSnapshot) { e.ActorUid = "not-a-uuid" }, want: codes.InvalidArgument}, + {name: "no expectation", phase: "unconditional", mutate: func(a *ateapi.Actor) { local(a); pending(a) }}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + st := newTestPersistence(t) + newTestActorWorkflow(t, st, "ns", "tmpl1") + if tc.apiUpdate { + newTestActorWorkflow(t, st, "ns", "tmpl2") + } + seedRuntimeWorker(t, ctx, st, "pod-1", "node-1", testSandboxRuntime()) + w, atelet, _ := newWireFinalizeWorkflow(t, st) + scheduling := newSchedulingWorkflow(t, ctx, st) + w.scheduler, w.workerCache = scheduling.scheduler, scheduling.workerCache + actor := storetest.MustCreateActor(t, ctx, st, &ateapi.Actor{ + Metadata: &ateapi.ResourceMetadata{Atespace: "team-a", Name: "id1"}, + ActorTemplate: &ateapi.ObjectRef{Atespace: "ns", Name: "tmpl1"}, + Status: &ateapi.ActorStatus{State: ateapi.ActorState_ACTOR_STATE_SUSPENDED, + LastAssignedGeneration: 1, Snapshots: []*ateapi.Snapshot{durableMemorySnapshot(t, testSandboxRuntime())}}, + }) + ref := resources.ActorRefFromActor(actor) + s := &RPCService{impl: st, actorWorkflow: w} + observed, err := s.GetActor(ctx, &ateapi.GetActorRequest{Actor: ref.ToObjectRef()}) + if err != nil { + t.Fatal(err) + } + req := &ateapi.ResumeActorRequest{Actor: ref.ToObjectRef(), Expected: &ateapi.ExpectedSnapshot{ + ActorUid: observed.GetMetadata().GetUid(), SnapshotUuid: "snap-1", Generation: 1, + }} + if tc.request != nil { + tc.request(req.Expected) + } + injected := false + mutate := func() { + update := st.UpdateActor + if tc.apiUpdate { + update = (&ServiceImpl{store: st}).UpdateActor + } + actor, err = update(ctx, ref, store.PreconditionFrom(actor), func(a *ateapi.Actor) error { tc.mutate(a); return nil }) + if err != nil { + t.Fatal(err) + } + injected = true + } + counted := &leaseCountingStore{Interface: st} + switch tc.phase { + case "conflict": + counted.Interface = &conflictInjectingStore{Interface: st, inject: mutate} + case "lease": + counted.Interface = &resumeLeaseChangingStore{Interface: st, beforeAcquire: mutate} + case "resuming": + tmpl, err := resolveActorTemplate(ctx, st, actor) + if err != nil { + t.Fatal(err) + } + actor, _, err = w.ensureWorkerAssigned(ctx, ref, actor, tmpl, nil) + if err != nil || actor.GetStatus().GetState() != ateapi.ActorState_ACTOR_STATE_RESUMING { + t.Fatalf("persist RESUMING: %v, %v", actor, err) + } + case "suspend": + if _, err := s.ResumeActor(ctx, &ateapi.ResumeActorRequest{Actor: ref.ToObjectRef()}); err != nil { + t.Fatal(err) + } + if _, err := s.SuspendActor(ctx, &ateapi.SuspendActorRequest{Actor: ref.ToObjectRef(), Options: &ateapi.SuspendOptions{Mode: ateapi.DurabilityMode_DURABILITY_MODE_SYNCHRONOUS}}); err != nil { + t.Fatal(err) + } + actor = mustReadActor(t, ctx, st, ref) + if latest := snapshotAtLatestGeneration(actor.GetStatus()); latest.GetGeneration() != 2 || latest.GetUuid() == "snap-1" { + t.Fatalf("intervening suspend did not replace checkpoint: %v", latest) + } + case "replace": + if _, err := st.DeleteActor(ctx, ref, store.DeletePreconditions{}); err != nil { + t.Fatal(err) + } + actor.Metadata = &ateapi.ResourceMetadata{Atespace: ref.Atespace, Name: ref.Name} + actor = storetest.MustCreateActor(t, ctx, st, actor) + case "malformed": + // Invalid input must fail with no workflow or external dependencies. + s = &RPCService{} + default: + if tc.mutate != nil { + mutate() + } + if tc.phase == "unconditional" { + req.Expected = nil + } + } + w.store = counted + previousRestore, previousRun := atelet.requests() + resp, err := s.ResumeActor(ctx, req) + if got := apierror.Code(err); got != tc.want { + t.Fatalf("ResumeActor = %v (%v), want %v", got, err, tc.want) + } + wantLeases := 1 + if tc.phase == "malformed" { + wantLeases = 0 + } + if counted.acquireCalls != wantLeases { + t.Errorf("lease acquisitions = %d, want %d", counted.acquireCalls, wantLeases) + } + if (tc.phase == "conflict" || tc.phase == "lease") && !injected { + t.Error("concurrent update was not injected") + } + restore, run := atelet.requests() + after := mustReadActor(t, ctx, st, ref) + if tc.want != codes.OK { + if tc.diagnostic != "" { + var details []string + if tc.diagnostic == "actor UID" { + details = []string{after.GetMetadata().GetUid(), req.GetExpected().GetActorUid()} + } else if tc.diagnostic == "newest held snapshot" { + held := latestSnapshot(after.GetStatus(), func(*ateapi.Snapshot) bool { return true }) + details = []string{held.GetUuid(), fmt.Sprintf("generation %d", held.GetGeneration()), req.GetExpected().GetSnapshotUuid(), fmt.Sprintf("generation %d", req.GetExpected().GetGeneration())} + } + for _, detail := range append(details, tc.diagnostic, "expected") { + if !strings.Contains(err.Error(), detail) { + t.Errorf("rejection %q omits %q", err, detail) + } + } + } + if !proto.Equal(restore, previousRestore) || !proto.Equal(run, previousRun) || !proto.Equal(after, actor) { + t.Error("rejected resume called atelet or changed actor") + } + if assignment := firstAssignment(t, st, testWorkerUID("pod-1")); assignment != nil { + t.Errorf("rejected resume retained worker assignment: %v", assignment) + } + worker, err := st.GetWorker(ctx, testWorkerUID("pod-1")) + if err != nil || worker.GetStatus().GetAllocated().GetActors() != 0 { + t.Errorf("rejected resume retained capacity: worker=%v, err=%v", worker, err) + } + if tc.want != codes.Aborted { + return + } + // The same checkpoint can resume after reloading the new template. + resp, err = s.ResumeActor(ctx, req) + if err != nil { + t.Fatalf("identical retry after template change: %v", err) + } + if counted.acquireCalls != 2 { + t.Errorf("lease acquisitions after retry = %d, want 2", counted.acquireCalls) + } + after = mustReadActor(t, ctx, st, ref) + restore, run = atelet.requests() + } + if tc.noop { + if resp.GetResumed() || !proto.Equal(resp.GetActor(), actor) || !proto.Equal(after, actor) || !proto.Equal(restore, previousRestore) || !proto.Equal(run, previousRun) { + t.Errorf("already-running resume was not a no-op: %v", resp) + } + return + } + if !resp.GetResumed() || after.GetStatus().GetState() != ateapi.ActorState_ACTOR_STATE_RUNNING { + t.Errorf("resume did not finish: %v", resp) + } + if !proto.Equal(after.GetWorkerSelector(), actor.GetWorkerSelector()) { + t.Error("resume lost updated worker selector") + } + wantUUID, wantGeneration := "snap-1", int32(1) + if req.Expected != nil { + wantUUID, wantGeneration = req.Expected.GetSnapshotUuid(), req.Expected.GetGeneration() + } + if restore.GetSnapshot().GetSnapshotUuid() != wantUUID || restore.GetSnapshot().GetGeneration() != wantGeneration || run != nil { + t.Errorf("restore = %v, run = %v; want snapshot %q generation %d", restore, run, wantUUID, wantGeneration) + } + if tc.phase == "lost response" { + // The first response is lost; retry the original request unchanged. + atelet.mu.Lock() + atelet.restore, atelet.run = nil, nil + atelet.mu.Unlock() + retry, err := s.ResumeActor(ctx, req) + if err != nil { + t.Fatalf("identical retry after success: %v", err) + } + restore, run = atelet.requests() + if retry.GetResumed() || !proto.Equal(retry.GetActor(), after) || !proto.Equal(mustReadActor(t, ctx, st, ref), after) || restore != nil || run != nil { + t.Errorf("identical retry was not a no-op: %v", retry) + } + if counted.acquireCalls != 2 { + t.Errorf("lease acquisitions after retry = %d, want 2", counted.acquireCalls) + } + } + }) + } +} + +// resumeLeaseChangingStore changes the source after preflight, before the lease. +type resumeLeaseChangingStore struct { + store.Interface + beforeAcquire func() +} + +func (s *resumeLeaseChangingStore) AcquireLease(ctx context.Context, key string) (*store.Lease, error) { + s.beforeAcquire() + return s.Interface.AcquireLease(ctx, key) +} diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 1c81112384..8eb10d0188 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -100,7 +100,7 @@ func TestResumeActor_RunningFastPathDoesNotAcquireLease(t *testing.T) { st := &leaseCountingStore{Interface: persistence} w := &ActorWorkflow{store: st} - got, resumed, err := w.ResumeActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}) + got, resumed, err := w.ResumeActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, nil) if err != nil { t.Fatalf("ResumeActor: %v", err) } @@ -348,7 +348,7 @@ func TestAssignWorkerAttempt_BindNotFound(t *testing.T) { st := &bindErrorStore{Interface: persistence, err: tc.bindErr} w := &ActorWorkflow{store: st, workerCache: wc, scheduler: scheduling.New(wc)} - _, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) if tc.wantErr != nil { if !errors.Is(err, tc.wantErr) { t.Fatalf("assignWorkerAttempt error = %v, want %v", err, tc.wantErr) @@ -375,7 +375,7 @@ func TestEnsureWorkerAssigned_ConflictExhaustionIsRetryable(t *testing.T) { w := &ActorWorkflow{store: st, workerCache: wc, scheduler: scheduling.New(wc)} tmpl := &ateapi.ActorTemplate{} - _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl) + _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl, nil) if !errors.Is(err, store.ErrVersionConflict) { t.Fatalf("ensureWorkerAssigned error = %v, want ErrVersionConflict", err) } @@ -419,7 +419,7 @@ func TestAssignWorkerAttempt_SkipsWorkerAssignedInOtherAtespace(t *testing.T) { Metadata: &ateapi.ResourceMetadata{Atespace: "team-a", Name: "shared", Uid: "actor-uid"}, } tmpl := &ateapi.ActorTemplate{} - _, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "shared"}, actor, tmpl) + _, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "shared"}, actor, tmpl, nil) if apierror.Code(err) != codes.ResourceExhausted { t.Fatalf("assignWorkerAttempt() error = %v, want ResourceExhausted (no worker has room)", err) } @@ -476,7 +476,7 @@ func TestEnsureWorkerAssigned_MissingUploadClaimsNoWorker(t *testing.T) { } w := &ActorWorkflow{store: persistence, workerCache: wc, scheduler: scheduling.New(wc)} - if _, _, err := w.ensureWorkerAssigned(ctx, actorRef, actor, &ateapi.ActorTemplate{}); apierror.Code(err) != codes.DataLoss { + if _, _, err := w.ensureWorkerAssigned(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil); apierror.Code(err) != codes.DataLoss { t.Fatalf("ensureWorkerAssigned() error = %v, want DataLoss", err) } @@ -558,7 +558,7 @@ func TestAssignWorkerAttempt_RetryAfterConflictPicksFreshWorker(t *testing.T) { w := &ActorWorkflow{store: persistence, workerCache: wc, scheduler: scheduling.New(wc)} tmpl := &ateapi.ActorTemplate{} - _, worker, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl) + _, worker, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl, nil) if err != nil { t.Fatalf("assignWorkerAttempt() on retry = %v, want nil (must re-pick a free worker)", err) } @@ -706,7 +706,7 @@ func TestAssignWorkerAttempt_ConflictRefreshesActor(t *testing.T) { w := &ActorWorkflow{store: st, workerCache: wc, scheduler: scheduling.New(wc)} tmpl := &ateapi.ActorTemplate{} - refreshed, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl) + refreshed, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl, nil) if tc.wantRetry { if !errors.Is(err, store.ErrVersionConflict) { @@ -786,7 +786,7 @@ func TestResumeActorWorkflow_RejectedAndIdempotentPaths(t *testing.T) { } }) - actor, resumed, err := w.ResumeActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}) + actor, resumed, err := w.ResumeActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, nil) if tc.wantErr { if got := apierror.Code(err); got != codes.FailedPrecondition { t.Fatalf("apierror.Code(err) = %v, want %v (err: %v)", got, codes.FailedPrecondition, err) @@ -835,7 +835,7 @@ func TestEnsureWorkerAssigned_RejectsNonResumableStates(t *testing.T) { continue } actor := &ateapi.Actor{Status: &ateapi.ActorStatus{State: st}, Metadata: &ateapi.ResourceMetadata{Name: "id1", Uid: "actor-uid-1"}} - _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) assertPrerequisiteResult(t, st, err, false) } } @@ -871,7 +871,7 @@ func TestResumeActor_MetricSkipsAlreadyRunningNoop(t *testing.T) { } }) - _, _, err := w.ResumeActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}) + _, _, err := w.ResumeActor(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, nil) if tt.wantRecord && err == nil { t.Fatal("expected resume to fail, got nil error") } @@ -1081,7 +1081,7 @@ func TestLoadActorForResume_RunningActorShortCircuits(t *testing.T) { w := &ActorWorkflow{store: persistence} - actor, tmpl, err := w.loadActorForResume(ctx, actorRef) + actor, tmpl, err := w.loadActorForResume(ctx, actorRef, nil) if err != nil { t.Fatalf("loadActorForResume() unexpected error = %v", err) } @@ -1923,7 +1923,7 @@ func TestResumeActor_AteletWireRequest(t *testing.T) { a.Status.LastAssignedGeneration = gen }) - actor, loadedTmpl, err := w.loadActorForResume(ctx, actorRef) + actor, loadedTmpl, err := w.loadActorForResume(ctx, actorRef, nil) var tele restoreTelemetry if err == nil { tele, _, err = w.ensureAteletRestored(ctx, actorRef, actor, loadedTmpl, nil) @@ -2329,7 +2329,7 @@ func TestEnsureWorkerAssigned_ContentionRecordsOneObservation(t *testing.T) { } w, actor, reader := newContendedWorkflow(t, sched) - _, worker, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, worker, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) if err != nil { t.Fatalf("ensureWorkerAssigned: %v", err) } @@ -2363,7 +2363,7 @@ func TestEnsureWorkerAssigned_ExhaustedBackoffRecordsOneError(t *testing.T) { } w, actor, reader := newContendedWorkflow(t, sched) - _, _, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, _, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) if err == nil { t.Fatal("ensureWorkerAssigned succeeded, want a failure after the backoff ended") } @@ -2432,7 +2432,7 @@ func TestEnsureWorkerAssigned_CancelledDuringBackoffRecordsOne(t *testing.T) { } defer cancel() - _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) if !errors.Is(err, context.Canceled) { t.Fatalf("ensureWorkerAssigned error = %v, want context.Canceled", err) } @@ -2462,7 +2462,7 @@ func TestEnsureWorkerAssigned_CancelledBeforeFirstAttemptRecordsNothing(t *testi ctx, cancel := context.WithCancel(context.Background()) cancel() - _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) if !errors.Is(err, context.Canceled) { t.Fatalf("ensureWorkerAssigned error = %v, want context.Canceled", err) } @@ -2494,7 +2494,7 @@ func TestEnsureWorkerAssigned_CancelledInLastSleepRecordsOne(t *testing.T) { } defer cancel() - _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, _, err := w.ensureWorkerAssigned(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) if !errors.Is(err, context.Canceled) { t.Fatalf("ensureWorkerAssigned error = %v, want context.Canceled", err) } @@ -2523,7 +2523,7 @@ func TestEnsureWorkerAssigned_ContextEndsInsideAttemptRecordsOne(t *testing.T) { } w, actor, reader := newContendedWorkflow(t, sched) - _, _, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + _, _, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}, nil) if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("ensureWorkerAssigned error = %v, want context.DeadlineExceeded", err) } @@ -2707,7 +2707,7 @@ func TestAssignWorkerAttempt_PrefersRuntimeMatchingWorkerAndFallsBack(t *testing seedRuntimeWorker(t, ctx, st, "amd-pod", "node-amd", amd64RT) actor := seedActorWithSnapshot(t, ctx, st, suspended, "", durableMemorySnapshot(t, amd64RT)) - stored, worker, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}) + stored, worker, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } @@ -2723,7 +2723,7 @@ func TestAssignWorkerAttempt_PrefersRuntimeMatchingWorkerAndFallsBack(t *testing seedRuntimeWorker(t, ctx, st, "amd-pod", "node-amd", newDefaultRT, oldRestorableRT) actor := seedActorWithSnapshot(t, ctx, st, suspended, "", durableMemorySnapshot(t, oldRestorableRT)) - stored, _, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}) + stored, _, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } @@ -2738,7 +2738,7 @@ func TestAssignWorkerAttempt_PrefersRuntimeMatchingWorkerAndFallsBack(t *testing seedRuntimeWorker(t, ctx, st, "arm-pod", "node-arm", arm64RT) actor := seedActorWithSnapshot(t, ctx, st, suspended, "", durableMemorySnapshot(t, amd64RT)) - stored, worker, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}) + stored, worker, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } @@ -2766,7 +2766,7 @@ func TestAssignWorkerAttempt_PrefersRuntimeMatchingWorkerAndFallsBack(t *testing snap.Fidelity = fidelity actor := seedActorWithSnapshot(t, ctx, st, suspended, "", snap) - _, _, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}) + _, _, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil) if apierror.Code(err) != codes.ResourceExhausted { t.Fatalf("assignWorkerAttempt with only a microvm worker = %v, want ResourceExhausted", err) } @@ -2783,7 +2783,7 @@ func TestAssignWorkerAttempt_PrefersRuntimeMatchingWorkerAndFallsBack(t *testing local.SandboxRuntime = amd64RT actor := seedActorWithSnapshot(t, ctx, st, suspended, "node-1", local) - _, _, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}) + _, _, err := newSchedulingWorkflow(t, ctx, st).assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil) if got := apierror.Code(err); got != codes.ResourceExhausted { t.Fatalf("apierror.Code(err) = %v, want %v (err: %v)", got, codes.ResourceExhausted, err) } @@ -2813,7 +2813,7 @@ func TestAssignWorkerAttempt_PrefersRuntimeMatchingWorkerAndFallsBack(t *testing actor := seedActorWithSnapshot(t, ctx, st, suspended, "", durableMemorySnapshot(t, oldRestorableRT)) w := &ActorWorkflow{store: st, scheduler: &relaxingTestScheduler{Scheduler: scheduling.New(nil), relaxedWorker: worker}} - stored, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}) + stored, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } @@ -2833,7 +2833,7 @@ func TestAssignWorkerAttempt_PrefersRuntimeMatchingWorkerAndFallsBack(t *testing actor := seedActorWithSnapshot(t, ctx, st, suspended, "", durableMemorySnapshot(t, oldRestorableRT)) w := &ActorWorkflow{store: st, scheduler: &relaxingTestScheduler{Scheduler: scheduling.New(nil), relaxedWorker: stale}} - stored, bound, err := w.assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}) + stored, bound, err := w.assignWorkerAttempt(ctx, actorRef, actor, &ateapi.ActorTemplate{}, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } @@ -3179,7 +3179,7 @@ func TestResume_ChecksMountedVolumesBeforeAttach(t *testing.T) { // u is declared but not mounted; o is not declared at all. tmpl := volumeTemplate([]string{"a", "u", "b"}, "a", "b") - assigned, worker, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl) + assigned, worker, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl, nil) if err != nil { t.Fatalf("assignWorkerAttempt: %v", err) } @@ -3448,7 +3448,7 @@ func TestAssignWorkerAttempt_WorkerWithoutNodeRejectsVolumes(t *testing.T) { w := newVolumeAssignWorkflow(persistence, wc, plugin) tmpl := volumeTemplate([]string{"vol"}, tt.mounted...) - _, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl) + _, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl, nil) stored, gerr := persistence.GetActor(ctx, actorRef) if gerr != nil { t.Fatalf("GetActor: %v", gerr) @@ -3486,13 +3486,13 @@ func TestAssignWorkerAttempt_RefusesChangedOverhead(t *testing.T) { w.store.(*contentionStore).workers[locked.GetMetadata().GetName()] = locked tmpl := &ateapi.ActorTemplate{Resources: &ateapi.ResourceRequirements{Limits: &ateapi.Resources{MemoryBytes: 256 << 20}}} - _, _, err := w.assignWorkerAttempt(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl) + _, _, err := w.assignWorkerAttempt(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl, nil) if !errors.Is(err, errWorkerFilledUp) { t.Fatalf("assignWorkerAttempt() = %v, want %v", err, errWorkerFilledUp) } w.store.(*contentionStore).workers[locked.GetMetadata().GetName()] = cached - if _, _, err := w.assignWorkerAttempt(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl); err != nil { + if _, _, err := w.assignWorkerAttempt(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl, nil); err != nil { t.Fatalf("assignWorkerAttempt() with an unchanged overhead = %v", err) } } diff --git a/docs/reference/api-guide.md b/docs/reference/api-guide.md index e8baa8c878..a70571e006 100644 --- a/docs/reference/api-guide.md +++ b/docs/reference/api-guide.md @@ -416,7 +416,7 @@ Each entry in `containers` may declare an optional **HTTP wakeup probe** so `Res How it behaves: -- **When it runs.** On every `ResumeActor` that actually wakes the actor. A `ResumeActor` on an actor that is already `RUNNING` is a no-op and does not probe. +- **When it runs.** On every `ResumeActor` that actually wakes the actor. A successful `ResumeActor` on an actor that is already `RUNNING` is a no-op and does not probe. - **Block-until-ready semantics.** `ResumeActor` returns successfully only after every container with a `wakeupProbe` has returned HTTP 200. If any container has not done so within `timeoutSeconds` (30s by default), `ResumeActor` fails and the actor moves to `ACTOR_STATE_CRASHED`, as it does for any other error while waking. A crashed actor cannot be resumed directly: `ResumeActor` rejects it with `FAILED_PRECONDITION`. Call `RevertActor` first, which returns it to `ACTOR_STATE_SUSPENDED` with its last snapshot, then resume it again. - **Aggressive polling.** The poll loop is tuned for single-millisecond detection latency: a keep-alive HTTP client with a 1ms interval and 250ms per-request timeout. While the workload is still booting, kernel `RST`s return in microseconds, so the loop spends almost no time blocked; once the listener is up, the next attempt completes on veth-local latency. - **Egress while waking.** The actor's egress is active before its containers start, so a workload can download what it needs to become ready (a model, a Git repository) before it answers the probe. A microVM restored from a memory snapshot is already past its boot, and gets egress just before the probe starts. Its EgressPolicy applies as it does once the actor is `RUNNING`; inbound traffic still waits for the probe. A wake that fails or is cut short after egress is active can be retried from the same snapshot, so outbound requests made while waking can be sent more than once. @@ -454,6 +454,25 @@ resources: memoryBytes: 1073741824 # 1Gi ``` +### Conditional Resume + +To resume a checkpoint recorded by a client, set `ResumeActorRequest.expected` with +`actorUid`, `snapshotUuid`, and `generation` from the observed actor and snapshot. +The actor's UID and newest held snapshot must match. An already `RUNNING` actor +returns `resumed=false`, so retrying after a lost response is safe. A `SUSPENDED` +or retryable `RESUMING` actor resumes only if the snapshot it would restore also +matches and its durable copy is `COMPLETED`, even if a local copy exists. +If the snapshot's node was deleted before reporting completion, resume attempts +that snapshot's upload on another node. A successful restore records it as +`COMPLETED`; a missing upload moves the actor to `CRASHED` and returns +`FAILED_PRECONDITION` with the held and expected snapshot identities. Call +`RevertActor` before resuming again. A matching pending upload on a node not known to +be deleted returns `UNAVAILABLE`. A mismatched actor or checkpoint, +or an invalid state, returns `FAILED_PRECONDITION` because the caller's view is +stale. A concurrent template change returns `ABORTED`; retry the same request. +The condition does not select an older checkpoint or change restore fidelity. +Omit `expected` to retain unconditional resume behavior. + ### Snapshot Storage Layout `snapshotConfig.storageLocation` is a **base prefix**, not the address of any one snapshot. Every external snapshot has exactly one owner: the actor that took it, or the tag that copied it. And the owner is part of the path, so an object's name says who it belongs to: @@ -761,7 +780,8 @@ Because the guards are required and only a read supplies them, an update is alwa Activates a suspended actor by restoring it onto a physical worker. * **Request:** `ResumeActorRequest` * `actor`: `ObjectRef` of the actor to resume. -* **Response:** `ResumeActorResponse` containing the updated `Actor` object (including the physical worker placement in `status.worker_assignment`). + * `expected`: (Optional) `ExpectedSnapshot` with the observed `actorUid`, `snapshotUuid`, and `generation`. See [Conditional Resume](#conditional-resume). +* **Response:** `ResumeActorResponse` containing the updated `Actor` object (including the physical worker placement in `status.worker_assignment`) and `resumed`, which is false if the actor was already `RUNNING`. * **Errors:** `RESOURCE_EXHAUSTED` if no eligible worker has room. `FAILED_PRECONDITION` if the actor is `ACTOR_STATE_CRASHED`. When a suspended actor's node is deleted, resume restores its local snapshot on another node from the snapshot's upload, even if atelet never reported the upload complete. Resume does this once another node has room. If atelet finds no upload, resume moves the actor to `ACTOR_STATE_CRASHED` and returns `FAILED_PRECONDITION`. `RevertActor` then returns it to its last completed external snapshot. Revert deletes an upload that was never reported complete, so a snapshot is lost if its restore fails for another reason, or if the actor is reverted before it resumes. #### `SuspendActor`