diff --git a/cmd/ateapi/internal/controlapi/snapshot_report.go b/cmd/ateapi/internal/controlapi/snapshot_report.go index 4b5ef70f28..fa0db2e062 100644 --- a/cmd/ateapi/internal/controlapi/snapshot_report.go +++ b/cmd/ateapi/internal/controlapi/snapshot_report.go @@ -37,9 +37,9 @@ func (s *RPCService) ApplyActorSnapshotState(ctx context.Context, actorRef resou } // ApplyReportedSnapshotState applies a report from the atelet on node to the -// Actor. It holds the Actor's lease and applies reports only to a SUSPENDED or -// RUNNING Actor, so a report never interleaves with a workflow. A report that -// changes nothing writes nothing. +// Actor. It holds the Actor's lease and applies reports only to a SUSPENDED, +// RUNNING, or CRASHED Actor, so a report never interleaves with a workflow. A +// report that changes nothing writes nothing. func (w *ActorWorkflow) ApplyReportedSnapshotState(ctx context.Context, actorRef resources.ActorRef, actorUID, node string, state *ateapipb.ActorSnapshotState) error { ctx = preview.ContextWithOptIn(ctx) // Check before taking the lease, so an atelet can't hold the lease of an @@ -57,23 +57,26 @@ func (w *ActorWorkflow) ApplyReportedSnapshotState(ctx context.Context, actorRef return err } + applied := proto.Clone(before).(*ateapipb.Actor) // A workflow owns a transitioning Actor's snapshots, even after its lease - // lapses. A crashed Actor's snapshots wait for a revert. + // lapses. switch st := before.GetStatus().GetState(); st { case ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ateapipb.ActorState_ACTOR_STATE_RUNNING: + if err := applyActorSnapshotState(leaseCtx, applied.GetStatus(), node, state); err != nil { + return err + } case ateapipb.ActorState_ACTOR_STATE_CRASHED: - return apierror.FailedPrecondition("actor %s is %s; revert it first", actorRef, st) + // An Actor whose later generation crashed may have another async + // upload that uploaded its snapshot. + if err := applyUploadOnNode(leaseCtx, actorRef, applied.GetStatus(), node, state.GetLatestDurable()); err != nil { + return err + } case ateapipb.ActorState_ACTOR_STATE_SUSPENDING, ateapipb.ActorState_ACTOR_STATE_RESUMING, ateapipb.ActorState_ACTOR_STATE_DELETING, ateapipb.ActorState_ACTOR_STATE_REVERTING: return apierror.Aborted("actor %s is %s; retry once that finishes", actorRef, st) default: return apierror.Internal("actor %s has unexpected state %s", actorRef, st) } - - applied := proto.Clone(before).(*ateapipb.Actor) - if err := applyActorSnapshotState(leaseCtx, applied.GetStatus(), node, state); err != nil { - return err - } if proto.Equal(applied.GetStatus(), before.GetStatus()) { return nil } @@ -162,6 +165,28 @@ func applyAteletResponseState(ctx context.Context, status *ateapipb.ActorStatus, return applyLatestDurable(ctx, status, snapshotReportFromAtelet(state.GetLatestDurable())) } +// applyUploadOnNode marks the reported durable snapshot COMPLETED, as +// applyLatestDurable does, if the record places that snapshot on node. Only a +// finalized suspend places a snapshot on a node, so any other snapshot, such +// as one a crashed suspend created, stays PENDING and a revert deletes it with +// the rest of the discarded execution. +func applyUploadOnNode(ctx context.Context, actorRef resources.ActorRef, status *ateapipb.ActorStatus, node string, reported *ateapipb.SnapshotReport) error { + if reported == nil { + return nil + } + snap := findSnapshotByGeneration(status.GetSnapshots(), reported.GetGeneration()) + if got := snap.GetLocality(); got != node { + slog.InfoContext(ctx, "Discarding an upload reported for a snapshot not on its node", + slog.Any("actor", actorRef), + slog.String("node", node), + slog.String("snapshot_locality", got), + slog.Int("generation", int(reported.GetGeneration())), + slog.String("uuid", reported.GetUuid())) + return nil + } + return applyLatestDurable(ctx, status, reported) +} + // applyLatestDurable marks the reported durable snapshot COMPLETED, and records // its fidelity. Its uuid and upload URI must match the recorded ones, and its // fidelity must be set. The snapshot keeps its locality, so an Actor with a diff --git a/cmd/ateapi/internal/controlapi/snapshot_report_test.go b/cmd/ateapi/internal/controlapi/snapshot_report_test.go index 435244ff7c..054408c0e1 100644 --- a/cmd/ateapi/internal/controlapi/snapshot_report_test.go +++ b/cmd/ateapi/internal/controlapi/snapshot_report_test.go @@ -468,7 +468,7 @@ func TestApplyReportedSnapshotState(t *testing.T) { assertUnwritten(t) }) } - // A transitioning actor gets Aborted, and a crashed one FailedPrecondition. + // A transitioning actor gets Aborted. for _, tc := range []struct { state ateapipb.ActorState want codes.Code @@ -477,7 +477,6 @@ func TestApplyReportedSnapshotState(t *testing.T) { {ateapipb.ActorState_ACTOR_STATE_RESUMING, codes.Aborted}, {ateapipb.ActorState_ACTOR_STATE_DELETING, codes.Aborted}, {ateapipb.ActorState_ACTOR_STATE_REVERTING, codes.Aborted}, - {ateapipb.ActorState_ACTOR_STATE_CRASHED, codes.FailedPrecondition}, } { t.Run("refused for a "+tc.state.String()+" actor", func(t *testing.T) { actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { s.State = tc.state }) @@ -519,6 +518,73 @@ func TestApplyReportedSnapshotState(t *testing.T) { assertUnwritten(t) } +// A report for a CRASHED actor records only the upload, and only of a snapshot +// the record places on the reporting node. atelet forgets the actor when a +// revert or delete terminates it, so the report is the upload's last chance +// to be recorded before the revert deletes it as in progress. +func TestApplyReportedSnapshotState_Crashed(t *testing.T) { + completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED + pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING + for _, tc := range []struct { + name string + // locality is the snapshot's node. A suspend that crashed before it + // committed leaves its snapshot on no node. + locality string + // detach sends the report atelet sends once it detached the actor: + // no local snapshot and no upload in progress. + detach bool + wantDurable ateapipb.SnapshotStorageStatus + }{ + {name: "records the upload", locality: reportNode, wantDurable: completed}, + {name: "does not apply a detach", locality: reportNode, detach: true, wantDurable: completed}, + {name: "leaves a crashed suspend's upload pending", wantDurable: pending}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + persistence := newTestPersistence(t) + w := NewActorWorkflow(persistence, nil, nil, nil, nil, "", nil, nil) + actorRef := resources.ActorRef{Atespace: reportAtespace, Name: "actor-1"} + actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, + ActorTemplate: &ateapipb.ObjectRef{Atespace: reportAtespace, Name: "tmpl"}, + Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_CRASHED}, + }) + uid := actor.GetMetadata().GetUid() + uri := ownedURI(t, uid, 2) + actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { + s.AssignedNode = reportNode + s.LastAssignedGeneration = 2 + snap := newDurableSnapshot(2, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", snapName(2), uri, pending) + snap.Locality = tc.locality + s.Snapshots = []*ateapipb.Snapshot{snap} + }) + uploaded := reportedAt(t, 2, uri) + state := &ateapipb.ActorSnapshotState{LatestLocal: uploaded, LatestDurable: uploaded} + if tc.detach { + state.LatestLocal = nil + } + + if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, state); err != nil { + t.Fatalf("ApplyReportedSnapshotState() failed: %v", err) + } + got, err := persistence.GetActor(ctx, actorRef) + if err != nil { + t.Fatal(err) + } + snap := got.GetStatus().GetSnapshots()[0] + if st := snap.GetDurableSnapshot().GetStatus(); st != tc.wantDurable { + t.Errorf("durable snapshot status = %v, want %v", st, tc.wantDurable) + } + if snap.GetLocality() != tc.locality || got.GetStatus().GetAssignedNode() != reportNode { + t.Errorf("snapshot locality %q, assigned node %q, want %q and %q", snap.GetLocality(), got.GetStatus().GetAssignedNode(), tc.locality, reportNode) + } + if got.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_CRASHED { + t.Errorf("state = %v, want CRASHED", got.GetStatus().GetState()) + } + }) + } +} + func TestActorOnNode(t *testing.T) { actor := func(assigned string, localities ...string) *ateapipb.Actor { status := &ateapipb.ActorStatus{AssignedNode: assigned} diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index 070f1a2966..33f7afc041 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -61,7 +61,7 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto } var atletTerminatedErr, volumesDetachedErr error - if err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete); err != nil { + if _, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete); err != nil { atletTerminatedErr = fmt.Errorf("while terminating atelet: %w", err) errs = append(errs, atletTerminatedErr) } @@ -96,15 +96,17 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto return w.finalizeDeleted(ctx, actor) } -// ensureAteletTerminated calls atelet to terminate the workload for opName. -func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, opName string) (err error) { +// ensureAteletTerminated calls atelet to terminate the workload for opName. It +// returns the actor's state as atelet forgot it, or nil when it skipped the +// call. +func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, opName string) (_ *ateletpb.ActorSnapshotState, err error) { ctx, done := stepSpan(ctx, "CallAteletTerminate") defer func() { err = done(err) }() nodeName := actor.GetStatus().GetAssignedNode() if nodeName == "" { slog.InfoContext(ctx, "actor has no assigned node, skipping atelet terminate request", slog.Any("actor", actorRef)) - return nil + return nil, nil } workerPodUID := "" @@ -116,7 +118,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res // ordinary case, and the others are none of this delete's business. hosted, err := workerHostsActor(ctx, w.store, workerName, actor.GetMetadata().GetUid()) if err != nil { - return err + return nil, err } if !hosted { slog.InfoContext(ctx, "worker is no longer assigned to this actor, skipping worker workload termination", @@ -131,18 +133,18 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res if errors.Is(err, ErrNoAteletOnNode) { gone, nodeErr := w.dialer.NodeGone(ctx, nodeName) if nodeErr != nil { - return fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr) + return nil, fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr) } if gone { // A deleted Node object counts as the node being gone for good. If the // machine still runs, whatever atelet holds for the actor stays there. slog.WarnContext(ctx, "assigned node no longer exists, skipping atelet terminate request", slog.String("node", nodeName), slog.Any("actor", actorRef)) - return nil + return nil, nil } } if err != nil { - return fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err) + return nil, fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err) } client := ateletpb.NewAteletClient(conn) @@ -151,7 +153,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res if actorTemplate != nil { spec, err := workloadSpecFromActorTemplate(actorTemplate, actor, nil) if err != nil { - return err + return nil, err } workloadSpec = spec } else { @@ -174,15 +176,15 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res Spec: workloadSpec, } - if _, err := client.Terminate(ctx, req); err != nil { + resp, err := client.Terminate(ctx, req) + if err != nil { if status.Code(err) == codes.NotFound { slog.InfoContext(ctx, "workload already terminated on atelet", slog.Any("actor", actorRef)) - return nil + return nil, nil } - return handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err) + return nil, handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err) } - - return nil + return resp.GetState(), nil } // fallbackWorkloadSpec is the spec to terminate an actor with when its template diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index 76bb5ff02f..31d6a11cca 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -124,7 +124,7 @@ func TestEnsureAteletTerminated_NodeGone(t *testing.T) { w := newTestActorWorkflow(t, newTestPersistence(t), "team-a", "tmpl") w.dialer = NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "", WithNodeGetter(nodeGetter(tc.lookupErr, tc.nodes...))) - err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert") + _, err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert") if tc.wantErr != (err != nil) { t.Fatalf("ensureAteletTerminated = %v, want error %v", err, tc.wantErr) } diff --git a/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go b/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go index 28eb5c2ec8..17db669978 100644 --- a/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go +++ b/cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go @@ -194,7 +194,7 @@ func (w *WorkerWorkflow) releaseEarlierAssignment(ctx context.Context, worker *a return nil } if actor.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDED { - if err := w.crashBoundActor(ctx, worker, actorRef, actor, "Releasing actor from a restarted worker", crashMessageWorkerRestarted); err != nil { + if err := w.crashBoundActor(ctx, worker, actorRef, actor, nil, "Releasing actor from a restarted worker", crashMessageWorkerRestarted); err != nil { return err } } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_detach.go b/cmd/ateapi/internal/controlapi/workflow_resume_detach.go index 665ed311d5..4800b5fc55 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_detach.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_detach.go @@ -139,9 +139,6 @@ func (w *ActorWorkflow) recordDetach(ctx context.Context, actorRef resources.Act if err := applyDetach(applied, node, applied.GetLastAssignedGeneration()); err != nil { return nil, err } - if applied.GetState() == ateapipb.ActorState_ACTOR_STATE_REVERTING && applied.GetWorkerAssignment() == nil { - applied.AssignedNode = "" - } stored, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error { // The precondition makes toUpdate the version applied was built from. toUpdate.Status.Snapshots = applied.GetSnapshots() diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 21a5c5e64c..0ed1d1957c 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -1150,7 +1150,9 @@ type capturingAtelet struct { // onDetach, when set, runs before DetachActor returns. onDetach func() terminateErr error - onTerminate func() + // terminateState, when set, is the snapshot state Terminate returns. + // Terminate is unimplemented otherwise. + terminateState *ateletpb.ActorSnapshotState } func (f *capturingAtelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (*ateletpb.RestoreResponse, error) { @@ -1212,17 +1214,14 @@ func (f *capturingAtelet) DetachActor(ctx context.Context, req *ateletpb.DetachA func (f *capturingAtelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) { f.mu.Lock() - hook := f.onTerminate - f.mu.Unlock() - if hook != nil { - hook() - } - f.mu.Lock() - err := f.terminateErr + err, state := f.terminateErr, proto.CloneOf(f.terminateState) f.mu.Unlock() if err != nil { return nil, err } + if state != nil { + return &ateletpb.TerminateResponse{State: state}, nil + } return f.UnimplementedAteletServer.Terminate(ctx, req) } diff --git a/cmd/ateapi/internal/controlapi/workflow_revert.go b/cmd/ateapi/internal/controlapi/workflow_revert.go index 55c239fe25..cccb6d9809 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert.go @@ -26,16 +26,18 @@ import ( "go.ate.dev/substrate/internal/apierror" "go.ate.dev/substrate/internal/ateattr" "go.ate.dev/substrate/internal/preview" + "go.ate.dev/substrate/internal/proto/ateletpb" "go.ate.dev/substrate/internal/resources" "go.ate.dev/substrate/pkg/proto/ateapipb" "go.opentelemetry.io/otel/attribute" + "google.golang.org/protobuf/proto" ) // revertableStates are the states a revert is accepted from. A SUSPENDED // actor is also accepted when its latest snapshot is node-local: the revert // discards the node-local copy and returns the actor to its latest durable -// snapshot, which is that same snapshot once atelet's upload of it is -// recorded. See isRevertable. +// snapshot, which is that same snapshot once atelet has uploaded it. See +// isRevertable. var revertableStates = []ateapipb.ActorState{ ateapipb.ActorState_ACTOR_STATE_RUNNING, ateapipb.ActorState_ACTOR_STATE_CRASHED, @@ -91,9 +93,11 @@ func (w *ActorWorkflow) RevertActor(ctx context.Context, actorRef resources.Acto } actor = marked - if err = w.ensureWorkerDiscarded(leaseCtx, actorRef, actor, actorTemplate); err != nil { + var discarded *ateapipb.Actor + if discarded, err = w.ensureWorkerDiscarded(leaseCtx, actorRef, actor, actorTemplate); err != nil { return nil, err } + actor = discarded if err = w.ensureInProgressSnapshotDiscarded(leaseCtx, actor); err != nil { return nil, err @@ -181,34 +185,75 @@ func (w *ActorWorkflow) ensureMarkedReverting(ctx context.Context, actorRef reso // worker. An assignment commits before the actor is updated to point at it, so // a crash in between leaves a row nothing references — and a CRASHED actor is // the likeliest holder of one. Absence is the ordinary case, not an error. -func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (err error) { +// +// It returns the actor as stored after the upload Terminate reports is +// recorded. +func (w *ActorWorkflow) ensureWorkerDiscarded(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (_ *ateapipb.Actor, err error) { ctx, done := stepSpan(ctx, "DiscardWorker") defer func() { err = done(err) }() if actor.GetStatus().GetAssignedNode() != "" { - forgetNode := false - if err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert); err != nil { - if !canDetachWithoutAtelet(ctx, actor, err) { - return err - } - forgetNode = true + state, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationRevert) + // The revert can go on without an unreachable atelet when nothing runs + // on the node and its latest local snapshot is already uploaded. + // FinalizeReverted drops the node binding, and the actor resumes from + // the uploaded copy. + if err != nil && !canDetachWithoutAtelet(ctx, actor, err) { + return nil, err } - if err := w.ensureVolumesDetached(ctx, actor, actorTemplate, "DetachVolumesForRevert", ateattr.OperationRevert); err != nil { - return err + if actor, err = w.recordTerminatedUpload(ctx, actorRef, actor, state); err != nil { + return nil, err } - if forgetNode { - if _, err := w.recordDetach(ctx, actorRef, actor, actor.GetStatus().GetAssignedNode(), nil); err != nil { - return err - } + if err := w.ensureVolumesDetached(ctx, actor, actorTemplate, "DetachVolumesForRevert", ateattr.OperationRevert); err != nil { + return nil, err } if actor.GetStatus().GetWorkerAssignment() != nil { if _, err := releaseWorker(ctx, w.store, actor); err != nil { - return fmt.Errorf("while releasing worker: %w", err) + return nil, fmt.Errorf("while releasing worker: %w", err) } } } - return w.releaseAssignmentWithoutBacklink(ctx, actor) + return actor, w.releaseAssignmentWithoutBacklink(ctx, actor) +} + +// recordTerminatedUpload records the upload that Terminate reported in state, +// as applyTerminatedUpload does. atelet may upload a snapshot without its +// report landing, so without this the revert would delete the upload as in +// progress. It writes at once because atelet forgot the actor, so a retried +// Terminate returns nothing. +func (w *ActorWorkflow) recordTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, state *ateletpb.ActorSnapshotState) (*ateapipb.Actor, error) { + node := actor.GetStatus().GetAssignedNode() + applied := proto.CloneOf(actor.GetStatus()) + if err := applyTerminatedUpload(ctx, actorRef, applied, state); err != nil { + return nil, err + } + if proto.Equal(applied, actor.GetStatus()) { + return actor, nil + } + stored, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error { + // The precondition makes toUpdate the version applied was built from. + toUpdate.Status.Snapshots = applied.GetSnapshots() + return nil + }) + if err != nil { + if errors.Is(err, store.ErrVersionConflict) { + return nil, apierror.Aborted("concurrent update conflict, please retry") + } + return nil, fmt.Errorf("while recording the upload atelet on node %q reported for Actor %s: %w", node, actorRef, err) + } + return stored, nil +} + +// applyTerminatedUpload records in status the upload that the atelet on the +// actor's assigned node reported durable in state as it terminated the actor, +// as applyUploadOnNode does. +func applyTerminatedUpload(ctx context.Context, actorRef resources.ActorRef, status *ateapipb.ActorStatus, state *ateletpb.ActorSnapshotState) error { + node := status.GetAssignedNode() + if err := applyUploadOnNode(ctx, actorRef, status, node, snapshotReportFromAtelet(state.GetLatestDurable())); err != nil { + return apierror.Internal("atelet on node %q reported a durable snapshot for Actor %s that conflicts with its record: %w", node, actorRef, err) + } + return nil } // ensureInProgressSnapshotDiscarded removes pending uploads of the discarded execution. diff --git a/cmd/ateapi/internal/controlapi/workflow_revert_test.go b/cmd/ateapi/internal/controlapi/workflow_revert_test.go index 31635b8518..7c932856b5 100644 --- a/cmd/ateapi/internal/controlapi/workflow_revert_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_revert_test.go @@ -19,11 +19,11 @@ import ( "testing" "github.com/google/go-cmp/cmp" - "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/objectstore/objectstoretest" "go.ate.dev/substrate/internal/objectstoreplugin/objectstoreplugintest" + "go.ate.dev/substrate/internal/proto/ateletpb" "go.ate.dev/substrate/internal/resources" "go.ate.dev/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" @@ -231,7 +231,6 @@ func TestRevertActor_UnreachableNode(t *testing.T) { pending bool assigned bool crashed bool - conflict bool wantCode codes.Code }{ {name: "missing atelet", node: "node-3"}, @@ -241,7 +240,6 @@ func TestRevertActor_UnreachableNode(t *testing.T) { {name: "worker still assigned", node: "node-1", err: apierror.Unavailable("offline"), assigned: true, wantCode: codes.Unavailable}, {name: "crashed execution still needs termination", node: "node-1", err: apierror.Unavailable("offline"), crashed: true, wantCode: codes.Unavailable}, {name: "other RPC failure", node: "node-1", err: apierror.Internal("injected failure"), wantCode: codes.Internal}, - {name: "concurrent update", node: "node-1", err: apierror.Unavailable("offline"), conflict: true, wantCode: codes.Aborted}, } { t.Run(tt.name, func(t *testing.T) { ctx := context.Background() @@ -279,21 +277,6 @@ func TestRevertActor_UnreachableNode(t *testing.T) { }) want := proto.CloneOf(actor.GetStatus().GetSnapshots()[1]) atelet.terminateErr = tt.err - if tt.conflict { - atelet.onTerminate = func() { - cur, err := st.GetActor(ctx, ref) - if err != nil { - t.Errorf("GetActor: %v", err) - return - } - if _, err := st.UpdateActor(ctx, ref, store.PreconditionFrom(cur), func(a *ateapipb.Actor) error { - a.WorkerSelector = &ateapipb.Selector{MatchLabels: map[string]string{"changed": "true"}} - return nil - }); err != nil { - t.Errorf("UpdateActor: %v", err) - } - } - } reverted, err := w.RevertActor(ctx, ref) if got := apierror.Code(err); got != tt.wantCode { t.Fatalf("RevertActor = %v, want %v", err, tt.wantCode) @@ -310,16 +293,7 @@ func TestRevertActor_UnreachableNode(t *testing.T) { if !proto.Equal(got.GetSnapshots()[1], want) { t.Fatalf("snapshot changed after failed revert: %v", got.GetSnapshots()) } - if !tt.conflict { - return - } - atelet.mu.Lock() - atelet.onTerminate = nil - atelet.mu.Unlock() - reverted, err = w.RevertActor(ctx, ref) - if err != nil { - t.Fatalf("retry after conflict: %v", err) - } + return } got := reverted.GetStatus() if got.GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDED || got.GetAssignedNode() != "" || got.GetWorkerAssignment() != nil { @@ -336,56 +310,176 @@ func TestRevertActor_UnreachableNode(t *testing.T) { } } -// TestRevertActor_PausedSnapshotAlreadyUploaded verifies a revert of a -// suspended actor whose local snapshot atelet uploaded returns it to that -// upload, its latest durable snapshot, and collects the durable snapshot the -// upload replaced. The actor has no assigned node, so there is nothing to -// terminate, as in TestRevertActor_ReturnsActorToItsSnapshot. +// TestRevertActor_PausedSnapshotAlreadyUploaded verifies a revert returns the +// actor to a local snapshot atelet uploaded, its latest durable snapshot, and +// collects the durable snapshot the upload replaced. The revert learns of the +// upload from the record, or from Terminate when atelet's report has not +// landed. An upload atelet does not report, or of a snapshot not on the node +// Terminate ran on, is deleted as in progress. func TestRevertActor_PausedSnapshotAlreadyUploaded(t *testing.T) { memory := ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED - ctx := context.Background() - st, cleanup := storetest.SetupTestStore(t) - defer cleanup() - w := newTestActorWorkflow(t, st, "ns", "tmpl1") - objects := objectstoretest.New() - w.objectStorePlugin = objectstoreplugintest.ControlClient(objects) + pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING + for _, tc := range []struct { + name string + state ateapipb.ActorState + // recorded is the upload's status on the record. + recorded ateapipb.SnapshotStorageStatus + // node is the actor's assigned node, which the revert terminates on. + node string + // terminateReports makes Terminate return the upload as durable. + terminateReports bool + // locality is the upload's node. A suspend that crashed before it + // committed leaves its snapshot on no node. + locality string + wantUploadKept bool + }{ + {name: "upload recorded", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: completed, locality: "node-1", wantUploadKept: true}, + {name: "paused, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1", terminateReports: true, locality: "node-1", wantUploadKept: true}, + {name: "crashed, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, locality: "node-1", wantUploadKept: true}, + {name: "upload not reported", state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, recorded: pending, node: "node-1", locality: "node-1"}, + {name: "crashed suspend, upload reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true}, + {name: "upload on another node reported by terminate", state: ateapipb.ActorState_ACTOR_STATE_CRASHED, recorded: pending, node: "node-1", terminateReports: true, locality: "node-2"}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + st, cleanup := storetest.SetupTestStore(t) + defer cleanup() + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + objects := objectstoretest.New() + w.objectStorePlugin = objectstoreplugintest.ControlClient(objects) + wire, atelet := newWireCaptureWorkflow(t, st) + w.dialer = wire.dialer - actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} - seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_SUSPENDED) - actor, err := st.GetActor(ctx, actorRef) - if err != nil { - t.Fatalf("get seeded actor: %v", err) - } - // newTestActorWorkflow's template stores snapshots under gs://snapshots. - tmpl := &ateapipb.ActorTemplate{SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}} - old := mustActorSnapshotURI(t, tmpl, actor, "old") - uploaded := mustActorSnapshotURI(t, tmpl, actor, "pause-snap") - objects.PutSnapshot(t, old, "manifest.json") - objects.PutSnapshot(t, uploaded, "manifest.json") - mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapipb.ActorStatus) { - paused := newPausedSnapshot(t, 2, memory, "", "pause-snap", "node-1") - paused.DurableSnapshot = &ateapipb.SnapshotStorage{Status: completed, Object: &ateapipb.ObjectSnapshot{SnapshotUri: uploaded.String()}} - s.LastAssignedGeneration = 2 - s.Snapshots = []*ateapipb.Snapshot{ - newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, memory, "", old.Name(), old.String(), completed), - paused, - } - }) + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", tc.state) + actor, err := st.GetActor(ctx, actorRef) + if err != nil { + t.Fatalf("get seeded actor: %v", err) + } + // newTestActorWorkflow's template stores snapshots under gs://snapshots. + tmpl := &ateapipb.ActorTemplate{SnapshotConfig: &ateapipb.SnapshotConfig{StorageLocation: "gs://snapshots"}} + old := mustActorSnapshotURI(t, tmpl, actor, "old") + uploaded := mustActorSnapshotURI(t, tmpl, actor, "pause-snap") + objects.PutSnapshot(t, old, "manifest.json") + objects.PutSnapshot(t, uploaded, "manifest.json") + mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapipb.ActorStatus) { + paused := newPausedSnapshot(t, 2, memory, "", "pause-snap", tc.locality) + paused.DurableSnapshot = &ateapipb.SnapshotStorage{Status: tc.recorded, Object: &ateapipb.ObjectSnapshot{SnapshotUri: uploaded.String()}} + s.AssignedNode = tc.node + s.LastAssignedGeneration = 2 + s.Snapshots = []*ateapipb.Snapshot{ + newDurableSnapshot(1, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, memory, "", old.Name(), old.String(), completed), + paused, + } + if tc.state == ateapipb.ActorState_ACTOR_STATE_CRASHED { + op := "ResumeActor" + if tc.locality == "" { + op = "SuspendActor" + } + s.Crash = newActorCrash(op, "interrupted execution") + } + }) + atelet.terminateErr = apierror.NotFound("atelet holds nothing for the actor") + if tc.terminateReports { + atelet.terminateErr = nil + atelet.terminateState = &ateletpb.ActorSnapshotState{LatestDurable: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{ + SnapshotUuid: "pause-snap", + Generation: 2, + Fidelity: ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, + Object: &ateletpb.ObjectStorage{SnapshotUri: uploaded.String()}, + }}} + } - reverted, err := w.RevertActor(ctx, actorRef) - if err != nil { - t.Fatalf("RevertActor: %v", err) - } - want := []snapshotOutcome{{generation: 2, fidelity: memory, durable: completed}} - if diff := cmp.Diff(want, describeSnapshots(reverted.GetStatus()), cmp.AllowUnexported(snapshotOutcome{})); diff != "" { - t.Errorf("snapshots mismatch (-want +got):\n%s", diff) - } - if len(objects.Snapshot(t, old)) != 0 { - t.Error("the durable snapshot the upload replaced was not released") + reverted, err := w.RevertActor(ctx, actorRef) + if err != nil { + t.Fatalf("RevertActor: %v", err) + } + want := []snapshotOutcome{{generation: 1, fidelity: memory, durable: completed}} + if tc.wantUploadKept { + want = []snapshotOutcome{{generation: 2, fidelity: memory, durable: completed}} + } + if diff := cmp.Diff(want, describeSnapshots(reverted.GetStatus()), cmp.AllowUnexported(snapshotOutcome{})); diff != "" { + t.Errorf("snapshots mismatch (-want +got):\n%s", diff) + } + // A kept upload replaces the old snapshot, which the revert collects. + if got, want := len(objects.Snapshot(t, old)) != 0, !tc.wantUploadKept; got != want { + t.Errorf("old snapshot kept = %v, want %v", got, want) + } + if got := len(objects.Snapshot(t, uploaded)) != 0; got != tc.wantUploadKept { + t.Errorf("uploaded snapshot kept = %v, want %v", got, tc.wantUploadKept) + } + }) } - if len(objects.Snapshot(t, uploaded)) == 0 { - t.Error("the upload the revert returned the actor to was released") +} + +// TestRecordTerminatedUpload verifies a revert records the upload Terminate +// reports. A report the record already holds writes nothing. A report that +// conflicts with the record, or a write that races another, fails the revert. +func TestRecordTerminatedUpload(t *testing.T) { + completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED + pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING + for _, tc := range []struct { + name string + // uploaded seeds the snapshot's upload as already recorded. + uploaded bool + // uuid is the snapshot uuid Terminate reports. + uuid string + // stale passes the actor as read before another write landed. + stale bool + wantCode codes.Code + wantDurable ateapipb.SnapshotStorageStatus + wantWrite bool + }{ + {name: "records the upload", uuid: "pause-snap", wantDurable: completed, wantWrite: true}, + {name: "upload already recorded", uploaded: true, uuid: "pause-snap", wantDurable: completed}, + {name: "report conflicts with the record", uuid: "other-snap", wantCode: codes.Internal, wantDurable: pending}, + {name: "concurrent update", uuid: "pause-snap", stale: true, wantCode: codes.Aborted, wantDurable: pending}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + st, cleanup := storetest.SetupTestStore(t) + defer cleanup() + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + + ref := resources.ActorRef{Atespace: "team-a", Name: "id1"} + shape := withPausedShape(t, "node-1") + if tc.uploaded { + shape = withUploadedPausedShape(t, "node-1") + } + seedWorkflowActor(t, ctx, st, ref, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_REVERTING, shape) + actor, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatalf("get seeded actor: %v", err) + } + before := actor + if tc.stale { + before = mustUpdateActorStatus(t, ctx, st, actor, func(s *ateapipb.ActorStatus) { + s.Crash = newActorCrash("ResumeActor", "interrupted execution") + }) + } + state := &ateletpb.ActorSnapshotState{LatestDurable: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{ + SnapshotUuid: tc.uuid, + Generation: 1, + Fidelity: ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, + Object: &ateletpb.ObjectStorage{SnapshotUri: someActorSnapshotURI(t, testStorageLocation, "team-a", "pause-snap")}, + }}} + + _, err = w.recordTerminatedUpload(ctx, ref, actor, state) + if got := apierror.Code(err); got != tc.wantCode { + t.Fatalf("recordTerminatedUpload = %v, want %v", err, tc.wantCode) + } + stored, err := st.GetActor(ctx, ref) + if err != nil { + t.Fatalf("GetActor: %v", err) + } + if got := stored.GetStatus().GetSnapshots()[0].GetDurableSnapshot().GetStatus(); got != tc.wantDurable { + t.Errorf("snapshot durable status = %v, want %v", got, tc.wantDurable) + } + if got := stored.GetMetadata().GetVersion() != before.GetMetadata().GetVersion(); got != tc.wantWrite { + t.Errorf("wrote the actor = %v, want %v", got, tc.wantWrite) + } + }) } } diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go index 9513e68bee..ca58f2763d 100644 --- a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go @@ -30,6 +30,7 @@ import ( "go.ate.dev/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) // defaultWorkerReclaimTimeout bounds each Terminate a worker delete sends to @@ -233,18 +234,18 @@ func (w *WorkerWorkflow) releaseBoundActor(ctx context.Context, worker *ateapipb // crash keeps the actor's assigned node, so if this best-effort reclaim is // skipped or fails, deleting or reverting the actor reclaims the state // there later. - w.reclaimActorStateOnNode(ctx, worker, actor) + state := w.reclaimActorStateOnNode(ctx, worker, actor) - return w.crashBoundActor(ctx, worker, actorRef, actor, "Releasing actor from a worker whose pod is gone", crashMessageWorkerPodGone) + return w.crashBoundActor(ctx, worker, actorRef, actor, state, "Releasing actor from a worker whose pod is gone", crashMessageWorkerPodGone) } // crashBoundActor moves an Actor that lost its sandbox on the Worker to // ACTOR_STATE_CRASHED and clears its pointers at the Worker. crashMsg is -// recorded in the Actor's status. An Actor in ACTOR_STATE_DELETING is left -// untouched. The Actor's assignment row in the worker is left for the caller to -// release. +// recorded in the Actor's status, along with the upload Terminate reported in +// state, if any. An Actor in ACTOR_STATE_DELETING is left untouched. The +// Actor's assignment row in the worker is left for the caller to release. // A concurrent write to the Actor fails this as ABORTED. -func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.Worker, actorRef resources.ActorRef, actor *ateapipb.Actor, logMsg, crashMsg string) error { +func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.Worker, actorRef resources.ActorRef, actor *ateapipb.Actor, state *ateletpb.ActorSnapshotState, logMsg, crashMsg string) error { // DELETING is final. A CRASHED actor can be reverted and resumed, and the // delete may already have collected its volumes and snapshots. The delete's // retry clears the actor's pointers. @@ -266,6 +267,13 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.W wasAlreadyCrashed := actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_CRASHED + // atelet forgot the actor as it terminated it, so the crash records the + // upload Terminate reported. + applied := proto.CloneOf(actor.GetStatus()) + if err := applyTerminatedUpload(ctx, actorRef, applied, state); err != nil { + return err + } + // Snapshot crash attributes before pod and pool pointers are cleared on actor. sandboxClass := actor.GetStatus().GetWorkerAssignment().GetSandboxRuntime().GetSandboxClass() crashAttrs := ateattr.ActorMetricAttributes(actor, sandboxClass, opName) @@ -279,6 +287,8 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.W toUpdate.Status.Crash = newActorCrash(opName, crashMsg) } toUpdate.Status.WorkerAssignment = nil + // The precondition makes toUpdate the version applied was built from. + toUpdate.Status.Snapshots = applied.GetSnapshots() return nil }) switch { @@ -309,17 +319,18 @@ func (w *WorkerWorkflow) crashBoundActor(ctx context.Context, worker *ateapipb.W // // This call is best-effort so an unreachable node cannot wedge worker // deregistration. If it fails, the actor's revert or delete reclaims the state -// on its assigned node. Each Terminate gets w.reclaimTimeout so a hung unmount, +// on its assigned node. It returns the actor's state as Terminate returned it, +// or nil if Terminate was skipped or failed. Each Terminate gets w.reclaimTimeout so a hung unmount, // such as against a dead NFS server, cannot run out the delete's own deadline // and fail the release after it. -func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *ateapipb.Worker, actor *ateapipb.Actor) { +func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *ateapipb.Worker, actor *ateapipb.Actor) *ateletpb.ActorSnapshotState { ctx, done := stepSpan(ctx, "ReclaimActorStateOnNode") defer func() { _ = done(nil) }() nodeName := worker.GetNodeName() if w.nodeGaveUp(nodeName) { markSkipped(ctx, "an earlier reclaim on the node timed out or could not reach atelet") - return + return nil } actorRef := resources.ActorRefFromActor(actor) @@ -337,7 +348,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at w.giveUpOnNode(nodeName) slog.WarnContext(ctx, "Could not reach the atelet holding a released actor's state; leaving it, and other actors' state on the node, for their revert or delete", append(logAttrs, slog.Any("err", err))...) - return + return nil } slog.InfoContext(ctx, "Reclaiming the node state of an actor released from a worker whose pod is gone", logAttrs...) @@ -345,7 +356,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at // outlive it. The fallback spec carries the external volumes to unmount. callCtx, cancel := context.WithTimeout(ctx, w.reclaimTimeout) defer cancel() - _, err = ateletpb.NewAteletClient(conn).Terminate(callCtx, &ateletpb.TerminateRequest{ + resp, err := ateletpb.NewAteletClient(conn).Terminate(callCtx, &ateletpb.TerminateRequest{ WorkerPodUid: worker.GetWorkerPodUid(), Atespace: actor.GetMetadata().GetAtespace(), ActorName: actor.GetMetadata().GetName(), @@ -356,6 +367,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at }) switch status.Code(err) { case codes.OK: + return resp.GetState() case codes.NotFound: slog.InfoContext(ctx, "Actor already terminated on its node", logAttrs...) case codes.DeadlineExceeded, codes.Unavailable: @@ -366,6 +378,7 @@ func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *at slog.WarnContext(ctx, "Failed to reclaim a released actor's node state; leaving it for the actor's revert or delete", append(logAttrs, slog.Any("err", err))...) } + return nil } // finalizeDeleted removes the worker from the store and returns the deleted diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go index 0f29a746af..b09ad5c76e 100644 --- a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go @@ -437,8 +437,13 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) { // seed further shapes the stored actor. seed func(*ateapipb.Actor) // terminateErr is what the atelet answers, if anything. - terminateErr error - wantTerminate bool + terminateErr error + // terminateState is the snapshot state a successful Terminate returns. + terminateState *ateletpb.ActorSnapshotState + wantTerminate bool + // wantDurable is the durable status of the actor's snapshot after the + // crash, if it has one. + wantDurable ateapipb.SnapshotStorageStatus }{ { name: "running actor is reclaimed", @@ -453,6 +458,23 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) { state: ateapipb.ActorState_ACTOR_STATE_RUNNING, seed: withPausedShape(t, "node-1"), wantTerminate: true, + wantDurable: ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING, + }, + { + // atelet forgets the actor as it terminates it, so the crash + // records the upload Terminate reports, or a revert would delete + // it as in progress. + name: "upload terminate reports is recorded", + state: ateapipb.ActorState_ACTOR_STATE_RUNNING, + seed: withPausedShape(t, "node-1"), + terminateState: &ateletpb.ActorSnapshotState{LatestDurable: &ateletpb.SnapshotReport{Snapshot: &ateletpb.Snapshot{ + SnapshotUuid: "pause-snap", + Generation: 1, + Fidelity: ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, + Object: &ateletpb.ObjectStorage{SnapshotUri: someActorSnapshotURI(t, testStorageLocation, "team-a", "pause-snap")}, + }}}, + wantTerminate: true, + wantDurable: ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED, }, { // A cleanly suspended actor already ran Checkpoint's teardown, @@ -478,7 +500,7 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) { persistence, cleanup := storetest.SetupTestStore(t) t.Cleanup(cleanup) - atelet := &capturingTerminator{err: tt.terminateErr} + atelet := &capturingTerminator{err: tt.terminateErr, state: tt.terminateState} wf := NewWorkerWorkflow(persistence, newNodeAteletDialer(t, atelet)) seedAPIWorker(t, ctx, persistence, validWorker(apiWorkerName)) @@ -515,6 +537,18 @@ func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) { if want := validWorker(apiWorkerName).GetWorkerPodUid(); req.GetWorkerPodUid() != want { t.Errorf("Terminate worker_pod_uid = %q, want the worker's pod uid %q", req.GetWorkerPodUid(), want) } + + crashed, err := persistence.GetActor(ctx, apiActorRef) + if err != nil { + t.Fatalf("GetActor: %v", err) + } + var gotDurable ateapipb.SnapshotStorageStatus + if snaps := crashed.GetStatus().GetSnapshots(); len(snaps) > 0 { + gotDurable = snaps[0].GetDurableSnapshot().GetStatus() + } + if gotDurable != tt.wantDurable { + t.Errorf("snapshot durable status = %v, want %v", gotDurable, tt.wantDurable) + } }) } } @@ -725,21 +759,22 @@ func TestDeleteWorkerWorkflow_HungReclaimDoesNotWedgeDelete(t *testing.T) { } // capturingTerminator is an atelet that records the Terminate calls it is sent -// and answers them with err, or, with hang set, blocks until the caller gives -// up, like an atelet stuck in an unmount. +// and answers them with err, or with state if err is nil. With hang set, it +// blocks until the caller gives up, like an atelet stuck in an unmount. type capturingTerminator struct { ateletpb.UnimplementedAteletServer - mu sync.Mutex - err error - hang bool - reqs []*ateletpb.TerminateRequest + mu sync.Mutex + err error + state *ateletpb.ActorSnapshotState + hang bool + reqs []*ateletpb.TerminateRequest } func (f *capturingTerminator) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) { f.mu.Lock() f.reqs = append(f.reqs, proto.Clone(req).(*ateletpb.TerminateRequest)) - hang, err := f.hang, f.err + hang, err, state := f.hang, f.err, proto.CloneOf(f.state) f.mu.Unlock() if hang { <-ctx.Done() @@ -748,7 +783,7 @@ func (f *capturingTerminator) Terminate(ctx context.Context, req *ateletpb.Termi if err != nil { return nil, err } - return &ateletpb.TerminateResponse{}, nil + return &ateletpb.TerminateResponse{State: state}, nil } func (f *capturingTerminator) requests() []*ateletpb.TerminateRequest { diff --git a/cmd/atelet/internal/actordirs/actordirs.go b/cmd/atelet/internal/actordirs/actordirs.go index d005817f1a..84190d2222 100644 --- a/cmd/atelet/internal/actordirs/actordirs.go +++ b/cmd/atelet/internal/actordirs/actordirs.go @@ -146,9 +146,10 @@ func RemoveActorFilesExceptState(actorUID string) error { // It refuses to proceed while a volume directory is still populated, so a // failed unmount cannot become a deletion of the mount's contents, and removes // a bundle upper dir carrying an image's read-only modes without recreating -// directories. Terminate calls Forget first, which has already deleted -// snapshot-state.json; after a crash here, Start finds no state file, so only a -// retried Terminate finishes the cleanup. +// directories. Terminate calls Forget first, which leaves only the latest +// durable snapshot in snapshot-state.json. This deletes that file after the +// actor's other files, so a Terminate retried after a failure here returns the +// snapshot again. func RemoveActorDirs(actorUID string) error { if err := RemoveActorFilesExceptState(actorUID); err != nil { return err diff --git a/cmd/atelet/internal/snapshottracker/background_test.go b/cmd/atelet/internal/snapshottracker/background_test.go index 9eb2c356c4..b8591f1eb3 100644 --- a/cmd/atelet/internal/snapshottracker/background_test.go +++ b/cmd/atelet/internal/snapshottracker/background_test.go @@ -481,7 +481,7 @@ func TestForgetWaitsForInFlightOp(t *testing.T) { opCtx, op := beginOp(t, tr, s1) forgetDone := make(chan error, 1) - go func() { forgetDone <- tr.Forget(trackedActorUID) }() + go func() { _, err := tr.Forget(trackedActorUID); forgetDone <- err }() select { case <-opCtx.Done(): case <-time.After(10 * time.Second): @@ -557,7 +557,7 @@ func TestTrackerTerminateHoldsOffBackgroundWork(t *testing.T) { <-opCtx.Done() endOp(tr, op) }() - if err := tr.Forget(trackedActorUID); err != nil { + if _, err := tr.Forget(trackedActorUID); err != nil { t.Fatalf("Forget: %v", err) } } @@ -665,7 +665,7 @@ func TestTrackerBackoff(t *testing.T) { {"CommitDurable", func() { tr.CommitDurable(trackedActorUID, testRef("s2", 2, clk.Now())) }}, {"CommitPromoted", func() { tr.CommitPromoted(trackedActorUID, testRef("s3", 3, clk.Now())) }}, {"Forget", func() { - if err := tr.Forget(trackedActorUID); err != nil { + if _, err := tr.Forget(trackedActorUID); err != nil { t.Fatalf("Forget: %v", err) } }}, diff --git a/cmd/atelet/internal/snapshottracker/detach_test.go b/cmd/atelet/internal/snapshottracker/detach_test.go index 5b073f86d5..c4ce8e8cae 100644 --- a/cmd/atelet/internal/snapshottracker/detach_test.go +++ b/cmd/atelet/internal/snapshottracker/detach_test.go @@ -769,7 +769,8 @@ func TestTrackerForgetCancelsDetach(t *testing.T) { forgetDone := make(chan error, 1) go func() { - forgetDone <- tr.Forget(trackedActorUID) + _, err := tr.Forget(trackedActorUID) + forgetDone <- err }() if got := waitCanceled(t, store); got != "s1" { t.Fatalf("canceled detach upload = %q, want s1", got) diff --git a/cmd/atelet/internal/snapshottracker/lifecycle.go b/cmd/atelet/internal/snapshottracker/lifecycle.go index 8e9284002b..95edd6ceab 100644 --- a/cmd/atelet/internal/snapshottracker/lifecycle.go +++ b/cmd/atelet/internal/snapshottracker/lifecycle.go @@ -16,6 +16,7 @@ package snapshottracker import ( "context" + "fmt" "time" "go.ate.dev/substrate/internal/apierror" @@ -346,13 +347,18 @@ func (t *Tracker) holdLocked(actorUID string, a *actorSnapshots, kind holdKind) } } -// Forget drops the actor's state, in memory and on disk. It first cancels any -// background upload or detach of the actor in progress and waits for it to -// end, releasing the actor's lock meanwhile. -func (t *Tracker) Forget(actorUID string) error { +// Forget drops the actor's state and returns it with only its latest durable +// snapshot, since the caller deletes the actor's local snapshots. It leaves +// only that snapshot in the state file, for the caller to delete with the +// actor's directory, so a retry after a failed cleanup returns it again. It +// first cancels any background upload or detach of the actor in progress and +// waits for it to end, releasing the actor's lock meanwhile. +func (t *Tracker) Forget(actorUID string) (*ateletpb.ActorSnapshotState, error) { t.lock(actorUID) defer t.unlock(actorUID) - if a := t.findLocked(actorUID); a != nil { + state := &ateletpb.ActorSnapshotState{} + a := t.findLocked(actorUID) + if a != nil { t.addLocked(a) a.hold = holdTerminate for a.op != nil { @@ -362,12 +368,23 @@ func (t *Tracker) Forget(actorUID string) error { <-op.done t.lock(actorUID) } + state.LatestDurable = a.LatestDurable.report() } t.retries.Forget(actorUID) t.reporting.retries.Forget(actorUID) t.dropLocked(actorUID) + if a == nil || a.LatestDurable == nil { + t.undurableActorsTally.forget(actorUID) + return state, removeSnapshotState(actorUID) + } + // With nothing local and no report due, an atelet restarted before the + // caller's cleanup finishes starts no background work on the actor. + err := t.writeStateLocked(&actorSnapshots{actorUID: actorUID, actorSnapshotsRecord: actorSnapshotsRecord{LatestDurable: a.LatestDurable}}) t.undurableActorsTally.forget(actorUID) - return removeSnapshotState(actorUID) + if err != nil { + return state, fmt.Errorf("while saving the actor's latest durable snapshot: %w", err) + } + return state, nil } // setLatestLocalLocked records snap as the actor's latest local snapshot and diff --git a/cmd/atelet/internal/snapshottracker/metrics_test.go b/cmd/atelet/internal/snapshottracker/metrics_test.go index e0b65d303e..c377607ef0 100644 --- a/cmd/atelet/internal/snapshottracker/metrics_test.go +++ b/cmd/atelet/internal/snapshottracker/metrics_test.go @@ -441,7 +441,7 @@ func TestDetachDurationOutcomes(t *testing.T) { // Forget cancels the detach as its upload finishes, and waits for it // to end. forgetDone := make(chan error, 1) - go func() { forgetDone <- tr.Forget(trackedActorUID) }() + go func() { _, err := tr.Forget(trackedActorUID); forgetDone <- err }() waitCanceled(t, store) store.release <- nil returnsPromptly(t, "Forget", func() { @@ -698,7 +698,7 @@ func TestUndurableActorsInstrumentsDropActor(t *testing.T) { if got, want := readUndurable(t, reader), (undurableReading{age: 60}); got != want { t.Errorf("before Forget: %+v, want %+v", got, want) } - if err := tr.Forget(trackedActorUID); err != nil { + if _, err := tr.Forget(trackedActorUID); err != nil { t.Fatalf("Forget: %v", err) } if got, want := readUndurable(t, reader), (undurableReading{}); got != want { @@ -799,7 +799,7 @@ func TestUndurableActorsInstrumentsMultipleActors(t *testing.T) { t.Errorf("split across bands: %+v, want %+v", got, want) } - if err := tr.Forget(trackedActorUID); err != nil { + if _, err := tr.Forget(trackedActorUID); err != nil { t.Fatalf("Forget: %v", err) } if got, want := readUndurable(t, reader), (undurableReading{ diff --git a/cmd/atelet/internal/snapshottracker/report_test.go b/cmd/atelet/internal/snapshottracker/report_test.go index 65d378e181..2956fb9912 100644 --- a/cmd/atelet/internal/snapshottracker/report_test.go +++ b/cmd/atelet/internal/snapshottracker/report_test.go @@ -1213,7 +1213,7 @@ func TestTrackerForgetDuringReport(t *testing.T) { waitReported(t, reporter) returnsPromptly(t, "Forget during a report", func() { - if err := tr.Forget(trackedActorUID); err != nil { + if _, err := tr.Forget(trackedActorUID); err != nil { t.Errorf("Forget: %v", err) } }) @@ -1223,8 +1223,8 @@ func TestTrackerForgetDuringReport(t *testing.T) { t.Errorf("reported %v after Forget, want nothing", r.state) case <-time.After(50 * time.Millisecond): } - if _, err := os.Stat(ateletpath.ActorSnapshotStateFile(trackedActorUID)); !os.IsNotExist(err) { - t.Errorf("state file once the report ended: stat = %v, want none", err) + if got := newTracker(t).State(trackedActorUID); uuidOf(got.GetLatestDurable()) != "s1" || got.GetLatestLocal() != nil { + t.Errorf("state file once the report ended = %v, want only s1 durable", got) } } diff --git a/cmd/atelet/internal/snapshottracker/tracker_test.go b/cmd/atelet/internal/snapshottracker/tracker_test.go index 30114f1e20..ee8f58b2f1 100644 --- a/cmd/atelet/internal/snapshottracker/tracker_test.go +++ b/cmd/atelet/internal/snapshottracker/tracker_test.go @@ -474,14 +474,24 @@ func TestTrackerState(t *testing.T) { t.Errorf("State() on a fresh tracker = %v, want %v from the state file", got, st) } - if err := tr.Forget(trackedActorUID); err != nil { + forgotten, err := tr.Forget(trackedActorUID) + if err != nil { t.Fatalf("forget: %v", err) } - if _, err := os.Stat(ateletpath.ActorSnapshotStateFile(trackedActorUID)); !os.IsNotExist(err) { - t.Errorf("state file after forget: stat = %v, want it deleted", err) + want := &ateletpb.ActorSnapshotState{LatestDurable: st.GetLatestDurable()} + if !proto.Equal(forgotten, want) { + t.Errorf("Forget returned %v, want %v: s2 durable and nothing local", forgotten, want) } - if got := tr.State(trackedActorUID); !proto.Equal(got, &ateletpb.ActorSnapshotState{}) { - t.Errorf("state after forget = %v, want empty", got) + // Terminate deletes the state file after the actor's other files, so a + // Terminate retried after a failed cleanup forgets s2 again. + if got := newTracker(t).State(trackedActorUID); !proto.Equal(got, want) { + t.Errorf("State() on a fresh tracker after forget = %v, want %v from the state file", got, want) + } + if forgotten, err = tr.Forget(trackedActorUID); err != nil { + t.Fatalf("second forget: %v", err) + } + if !proto.Equal(forgotten, want) { + t.Errorf("second Forget returned %v, want %v", forgotten, want) } } diff --git a/cmd/atelet/internal/snapshottracker/upload_test.go b/cmd/atelet/internal/snapshottracker/upload_test.go index 863dd8dcfd..9347bf6e1b 100644 --- a/cmd/atelet/internal/snapshottracker/upload_test.go +++ b/cmd/atelet/internal/snapshottracker/upload_test.go @@ -346,21 +346,31 @@ func TestTrackerForgetCancelsUpload(t *testing.T) { done := startUpload(t, tr, store) tr.CommitLocal(trackedActorUID, testRef("s2", 2, tr.clock.Now())) - forgetDone := make(chan error, 1) - go func() { forgetDone <- tr.Forget(trackedActorUID) }() + type forgetResult struct { + state *ateletpb.ActorSnapshotState + err error + } + forgetDone := make(chan forgetResult, 1) + go func() { + state, err := tr.Forget(trackedActorUID) + forgetDone <- forgetResult{state, err} + }() if got := waitCanceled(t, store); got != "s1" { t.Fatalf("canceled upload = %q, want s1", got) } select { - case err := <-forgetDone: - t.Fatalf("Forget returned (%v) before the canceled upload ended", err) + case res := <-forgetDone: + t.Fatalf("Forget returned (%v) before the canceled upload ended", res.err) case <-time.After(50 * time.Millisecond): } store.release <- context.Canceled select { - case err := <-forgetDone: - if err != nil { - t.Errorf("Forget: %v", err) + case res := <-forgetDone: + if res.err != nil { + t.Errorf("Forget: %v", res.err) + } + if got := res.state.GetLatestDurable(); got != nil { + t.Errorf("Forget returned latest durable %v, want nil: the canceled upload did not finish", got) } case <-time.After(10 * time.Second): t.Fatal("timed out waiting for Forget to return") @@ -1048,7 +1058,7 @@ func TestTrackerTerminateHoldsOffSupersededUploadingCleanup(t *testing.T) { } if tc.forget { - if err := tr.Forget(trackedActorUID); err != nil { + if _, err := tr.Forget(trackedActorUID); err != nil { t.Fatalf("Forget: %v", err) } end() diff --git a/cmd/atelet/lifecycle_test.go b/cmd/atelet/lifecycle_test.go index 7bc276da1e..3a06de9cdd 100644 --- a/cmd/atelet/lifecycle_test.go +++ b/cmd/atelet/lifecycle_test.go @@ -1143,7 +1143,11 @@ func TestTerminateWithoutWorkerPodUID(t *testing.T) { Containers: []*ateletpb.Container{{Name: "app", Image: "example.com/app@sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"}}, } - if _, err := s.Terminate(ctx, &ateletpb.TerminateRequest{ + // atelet uploaded the snapshot, and the control plane has not recorded it. + uploaded := lifecycleSnapshot(actorUID, snapshotName, 1, ateletpb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY) + s.snapshots.CommitDurable(actorUID, snapshottracker.SnapshotRef{Snapshot: uploaded}) + + resp, err := s.Terminate(ctx, &ateletpb.TerminateRequest{ Atespace: atespace, ActorName: actorName, ActorUid: actorUID, @@ -1151,10 +1155,14 @@ func TestTerminateWithoutWorkerPodUID(t *testing.T) { ActorTemplateName: "counter", WorkerPodUid: "", Spec: spec, - }); err != nil { + }) + if err != nil { t.Fatalf("Terminate: %v", err) } + if got := resp.GetState().GetLatestDurable().GetSnapshot(); !proto.Equal(got, uploaded) { + t.Errorf("TerminateResponse.state.latest_durable = %v, want the uploaded snapshot %v", got, uploaded) + } if got := worker.actorDirs["TerminateWorkload"]; got != nil { t.Errorf("TerminateWorkload was called on worker (%v), want skipped when WorkerPodUid is empty", got) } @@ -1290,7 +1298,8 @@ func TestTerminateReclaimsStateWithoutALiveSandbox(t *testing.T) { // TestTerminateCanBeRetried pins that a repeated Terminate succeeds, both when // a single caller retries and when DeleteActor or RevertActor sends a second -// Terminate after DeleteWorker's best-effort Terminate. +// Terminate after DeleteWorker's best-effort Terminate. A retry after a failed +// Terminate still returns the actor's latest durable snapshot. func TestTerminateCanBeRetried(t *testing.T) { for _, tt := range []struct { name string @@ -1316,7 +1325,8 @@ func TestTerminateCanBeRetried(t *testing.T) { ctx := t.Context() snap := a.snapshot("pause-snap-1", 1) - if _, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil { + checkpointResp, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE) + if err != nil { t.Fatalf("Checkpoint: %v", err) } @@ -1341,7 +1351,7 @@ func TestTerminateCanBeRetried(t *testing.T) { WorkerPodUid: pluginWorkerPodUID, Spec: a.spec, } - _, err := a.s.Terminate(ctx, firstReq) + _, err = a.s.Terminate(ctx, firstReq) if tt.failFirst { if err == nil { t.Fatal("first Terminate succeeded through a populated volume directory") @@ -1358,9 +1368,17 @@ func TestTerminateCanBeRetried(t *testing.T) { if tt.secondWithoutWorker { secondReq.WorkerPodUid = "" } - if _, err := a.s.Terminate(ctx, secondReq); err != nil { + resp, err := a.s.Terminate(ctx, secondReq) + if err != nil { t.Fatalf("second Terminate: %v", err) } + var wantDurable *ateletpb.SnapshotReport + if tt.failFirst { + wantDurable = checkpointResp.GetState().GetLatestDurable() + } + if got := resp.GetState().GetLatestDurable(); !proto.Equal(got, wantDurable) { + t.Errorf("second Terminate returned latest_durable %v, want %v", got, wantDurable) + } if got := a.worker.terminations.Load(); got != tt.wantTeardowns { t.Errorf("worker saw %d TerminateWorkload calls, want %d", got, tt.wantTeardowns) } diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index d935c6df09..0f0cb2f97d 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -2005,8 +2005,9 @@ func (s *Atelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) // up the copies on any other NodeVmsWithLocalSnapshots. This is fine *as of // the day this was written* because today NodeVmsWithLocalSnapshots has at // most one item. - if err := s.snapshots.Forget(actorUID); err != nil { - return nil, fmt.Errorf("failed to delete snapshot state during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + state, err := s.snapshots.Forget(actorUID) + if err != nil { + return nil, fmt.Errorf("failed to forget snapshot state during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) } if err := snapshottracker.PruneLocalCheckpoints(ctx, actorUID); err != nil { return nil, fmt.Errorf("failed to prune local checkpoints during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) @@ -2018,7 +2019,7 @@ func (s *Atelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) } s.restoreFailures.reset(actorUID) - return &ateletpb.TerminateResponse{}, nil + return &ateletpb.TerminateResponse{State: state}, nil } // terminateWorkloadOnWorker tears the workload down on the worker hosting it. diff --git a/internal/proto/ateletpb/atelet.pb.go b/internal/proto/ateletpb/atelet.pb.go index 754fb1b093..d9c6387d43 100644 --- a/internal/proto/ateletpb/atelet.pb.go +++ b/internal/proto/ateletpb/atelet.pb.go @@ -1918,7 +1918,12 @@ func (x *TerminateRequest) GetSpec() *WorkloadSpec { } type TerminateResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` + state protoimpl.MessageState `protogen:"open.v1"` + // The actor's state as atelet forgets it: latest_local and uploading are + // unset, and latest_durable is its latest snapshot atelet knew was durable, + // if any. Terminate cancels an upload in progress, so that upload's snapshot + // is latest_durable only if the upload finished first. + State *ActorSnapshotState `protobuf:"bytes,1,opt,name=state,proto3" json:"state,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1953,6 +1958,13 @@ func (*TerminateResponse) Descriptor() ([]byte, []int) { return file_atelet_proto_rawDescGZIP(), []int{28} } +func (x *TerminateResponse) GetState() *ActorSnapshotState { + if x != nil { + return x.State + } + return nil +} + type RunRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // The UID of the worker pod whose the worker hosts the actor. @@ -4437,8 +4449,9 @@ const file_atelet_proto_rawDesc = "" + "\tactor_uid\x18\x04 \x01(\tR\bactorUid\x126\n" + "\x17actor_template_atespace\x18\x05 \x01(\tR\x15actorTemplateAtespace\x12.\n" + "\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12(\n" + - "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\"\x13\n" + - "\x11TerminateResponse\"\xf4\x03\n" + + "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\"E\n" + + "\x11TerminateResponse\x120\n" + + "\x05state\x18\x01 \x01(\v2\x1a.atelet.ActorSnapshotStateR\x05state\"\xf4\x03\n" + "\n" + "RunRequest\x12$\n" + "\x0eworker_pod_uid\x18\x01 \x01(\tR\fworkerPodUid\x12\x1a\n" + @@ -4733,84 +4746,85 @@ var file_atelet_proto_depIdxs = []int32{ 27, // 13: atelet.EgressDestinations.ports:type_name -> atelet.EgressPorts 28, // 14: atelet.EgressPorts.all:type_name -> atelet.EgressAllPorts 35, // 15: atelet.TerminateRequest.spec:type_name -> atelet.WorkloadSpec - 35, // 16: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec - 34, // 17: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway - 15, // 18: atelet.RunRequest.resources:type_name -> atelet.ResourceRequirements - 11, // 19: atelet.RunRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime - 46, // 20: atelet.WorkloadSpec.containers:type_name -> atelet.Container - 44, // 21: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume - 66, // 22: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry - 67, // 23: atelet.ExternalVolumeSource.publish_context:type_name -> atelet.ExternalVolumeSource.PublishContextEntry - 0, // 24: atelet.ActorMetadataItem.field:type_name -> atelet.ActorMetadataField - 39, // 25: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem - 40, // 26: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource - 41, // 27: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource - 42, // 28: atelet.SystemInfoVolumeSource.data_sources:type_name -> atelet.SystemInfoDataSource - 36, // 29: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolumeSource - 37, // 30: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource - 43, // 31: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolumeSource - 38, // 32: atelet.Volume.image:type_name -> atelet.ImageVolumeSource - 49, // 33: atelet.Container.env:type_name -> atelet.EnvVar - 50, // 34: atelet.Container.wakeup_probe:type_name -> atelet.WakeupProbe - 45, // 35: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount - 47, // 36: atelet.Container.security_context:type_name -> atelet.SecurityContext - 15, // 37: atelet.Container.resources:type_name -> atelet.ResourceRequirements - 48, // 38: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities - 51, // 39: atelet.WakeupProbe.http_get:type_name -> atelet.HttpGetAction - 2, // 40: atelet.ObjectStorage.sharing:type_name -> atelet.SnapshotSharing - 1, // 41: atelet.Snapshot.fidelity:type_name -> atelet.SnapshotFidelity - 53, // 42: atelet.Snapshot.object:type_name -> atelet.ObjectStorage - 54, // 43: atelet.SnapshotReport.snapshot:type_name -> atelet.Snapshot - 68, // 44: atelet.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp - 68, // 45: atelet.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp - 55, // 46: atelet.ActorSnapshotState.latest_local:type_name -> atelet.SnapshotReport - 55, // 47: atelet.ActorSnapshotState.uploading:type_name -> atelet.SnapshotReport - 55, // 48: atelet.ActorSnapshotState.latest_durable:type_name -> atelet.SnapshotReport - 68, // 49: atelet.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp - 3, // 50: atelet.UploadConfig.mode:type_name -> atelet.UploadMode - 35, // 51: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec - 54, // 52: atelet.CheckpointRequest.snapshot:type_name -> atelet.Snapshot - 57, // 53: atelet.CheckpointRequest.upload_config:type_name -> atelet.UploadConfig - 56, // 54: atelet.CheckpointResponse.state:type_name -> atelet.ActorSnapshotState - 35, // 55: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec - 54, // 56: atelet.RestoreRequest.snapshot:type_name -> atelet.Snapshot - 34, // 57: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway - 15, // 58: atelet.RestoreRequest.resources:type_name -> atelet.ResourceRequirements - 11, // 59: atelet.RestoreRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime - 54, // 60: atelet.PromoteSnapshotRequest.snapshot:type_name -> atelet.Snapshot - 56, // 61: atelet.PromoteSnapshotResponse.state:type_name -> atelet.ActorSnapshotState - 56, // 62: atelet.DetachActorResponse.state:type_name -> atelet.ActorSnapshotState - 29, // 63: atelet.WorkerSupport.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest - 4, // 64: atelet.WorkerSupport.AcquireWorkerEpoch:input_type -> atelet.AcquireWorkerEpochRequest - 6, // 65: atelet.WorkerSupport.RegisterWorker:input_type -> atelet.RegisterWorkerRequest - 7, // 66: atelet.WorkerSupport.SetWorkerSandboxRuntimes:input_type -> atelet.SetWorkerSandboxRuntimesRequest - 18, // 67: atelet.WorkerSupport.RequestActorSuspend:input_type -> atelet.RequestActorSuspendRequest - 22, // 68: atelet.WorkerSupport.GetActorEgressPolicy:input_type -> atelet.GetActorEgressPolicyRequest - 20, // 69: atelet.WorkerSupport.ReportActorCrashed:input_type -> atelet.ReportActorCrashedRequest - 33, // 70: atelet.Atelet.Run:input_type -> atelet.RunRequest - 58, // 71: atelet.Atelet.Checkpoint:input_type -> atelet.CheckpointRequest - 60, // 72: atelet.Atelet.Restore:input_type -> atelet.RestoreRequest - 62, // 73: atelet.Atelet.PromoteSnapshot:input_type -> atelet.PromoteSnapshotRequest - 64, // 74: atelet.Atelet.DetachActor:input_type -> atelet.DetachActorRequest - 31, // 75: atelet.Atelet.Terminate:input_type -> atelet.TerminateRequest - 30, // 76: atelet.WorkerSupport.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse - 5, // 77: atelet.WorkerSupport.AcquireWorkerEpoch:output_type -> atelet.AcquireWorkerEpochResponse - 17, // 78: atelet.WorkerSupport.RegisterWorker:output_type -> atelet.RegisterWorkerResponse - 8, // 79: atelet.WorkerSupport.SetWorkerSandboxRuntimes:output_type -> atelet.SetWorkerSandboxRuntimesResponse - 19, // 80: atelet.WorkerSupport.RequestActorSuspend:output_type -> atelet.RequestActorSuspendResponse - 23, // 81: atelet.WorkerSupport.GetActorEgressPolicy:output_type -> atelet.GetActorEgressPolicyResponse - 21, // 82: atelet.WorkerSupport.ReportActorCrashed:output_type -> atelet.ReportActorCrashedResponse - 52, // 83: atelet.Atelet.Run:output_type -> atelet.RunResponse - 59, // 84: atelet.Atelet.Checkpoint:output_type -> atelet.CheckpointResponse - 61, // 85: atelet.Atelet.Restore:output_type -> atelet.RestoreResponse - 63, // 86: atelet.Atelet.PromoteSnapshot:output_type -> atelet.PromoteSnapshotResponse - 65, // 87: atelet.Atelet.DetachActor:output_type -> atelet.DetachActorResponse - 32, // 88: atelet.Atelet.Terminate:output_type -> atelet.TerminateResponse - 76, // [76:89] is the sub-list for method output_type - 63, // [63:76] is the sub-list for method input_type - 63, // [63:63] is the sub-list for extension type_name - 63, // [63:63] is the sub-list for extension extendee - 0, // [0:63] is the sub-list for field type_name + 56, // 16: atelet.TerminateResponse.state:type_name -> atelet.ActorSnapshotState + 35, // 17: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec + 34, // 18: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway + 15, // 19: atelet.RunRequest.resources:type_name -> atelet.ResourceRequirements + 11, // 20: atelet.RunRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime + 46, // 21: atelet.WorkloadSpec.containers:type_name -> atelet.Container + 44, // 22: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume + 66, // 23: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry + 67, // 24: atelet.ExternalVolumeSource.publish_context:type_name -> atelet.ExternalVolumeSource.PublishContextEntry + 0, // 25: atelet.ActorMetadataItem.field:type_name -> atelet.ActorMetadataField + 39, // 26: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem + 40, // 27: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource + 41, // 28: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource + 42, // 29: atelet.SystemInfoVolumeSource.data_sources:type_name -> atelet.SystemInfoDataSource + 36, // 30: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolumeSource + 37, // 31: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource + 43, // 32: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolumeSource + 38, // 33: atelet.Volume.image:type_name -> atelet.ImageVolumeSource + 49, // 34: atelet.Container.env:type_name -> atelet.EnvVar + 50, // 35: atelet.Container.wakeup_probe:type_name -> atelet.WakeupProbe + 45, // 36: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount + 47, // 37: atelet.Container.security_context:type_name -> atelet.SecurityContext + 15, // 38: atelet.Container.resources:type_name -> atelet.ResourceRequirements + 48, // 39: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities + 51, // 40: atelet.WakeupProbe.http_get:type_name -> atelet.HttpGetAction + 2, // 41: atelet.ObjectStorage.sharing:type_name -> atelet.SnapshotSharing + 1, // 42: atelet.Snapshot.fidelity:type_name -> atelet.SnapshotFidelity + 53, // 43: atelet.Snapshot.object:type_name -> atelet.ObjectStorage + 54, // 44: atelet.SnapshotReport.snapshot:type_name -> atelet.Snapshot + 68, // 45: atelet.SnapshotReport.local_commit_time:type_name -> google.protobuf.Timestamp + 68, // 46: atelet.SnapshotReport.durable_complete_time:type_name -> google.protobuf.Timestamp + 55, // 47: atelet.ActorSnapshotState.latest_local:type_name -> atelet.SnapshotReport + 55, // 48: atelet.ActorSnapshotState.uploading:type_name -> atelet.SnapshotReport + 55, // 49: atelet.ActorSnapshotState.latest_durable:type_name -> atelet.SnapshotReport + 68, // 50: atelet.ActorSnapshotState.upload_due_time:type_name -> google.protobuf.Timestamp + 3, // 51: atelet.UploadConfig.mode:type_name -> atelet.UploadMode + 35, // 52: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec + 54, // 53: atelet.CheckpointRequest.snapshot:type_name -> atelet.Snapshot + 57, // 54: atelet.CheckpointRequest.upload_config:type_name -> atelet.UploadConfig + 56, // 55: atelet.CheckpointResponse.state:type_name -> atelet.ActorSnapshotState + 35, // 56: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec + 54, // 57: atelet.RestoreRequest.snapshot:type_name -> atelet.Snapshot + 34, // 58: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway + 15, // 59: atelet.RestoreRequest.resources:type_name -> atelet.ResourceRequirements + 11, // 60: atelet.RestoreRequest.sandbox_runtime:type_name -> atelet.ActorSandboxRuntime + 54, // 61: atelet.PromoteSnapshotRequest.snapshot:type_name -> atelet.Snapshot + 56, // 62: atelet.PromoteSnapshotResponse.state:type_name -> atelet.ActorSnapshotState + 56, // 63: atelet.DetachActorResponse.state:type_name -> atelet.ActorSnapshotState + 29, // 64: atelet.WorkerSupport.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest + 4, // 65: atelet.WorkerSupport.AcquireWorkerEpoch:input_type -> atelet.AcquireWorkerEpochRequest + 6, // 66: atelet.WorkerSupport.RegisterWorker:input_type -> atelet.RegisterWorkerRequest + 7, // 67: atelet.WorkerSupport.SetWorkerSandboxRuntimes:input_type -> atelet.SetWorkerSandboxRuntimesRequest + 18, // 68: atelet.WorkerSupport.RequestActorSuspend:input_type -> atelet.RequestActorSuspendRequest + 22, // 69: atelet.WorkerSupport.GetActorEgressPolicy:input_type -> atelet.GetActorEgressPolicyRequest + 20, // 70: atelet.WorkerSupport.ReportActorCrashed:input_type -> atelet.ReportActorCrashedRequest + 33, // 71: atelet.Atelet.Run:input_type -> atelet.RunRequest + 58, // 72: atelet.Atelet.Checkpoint:input_type -> atelet.CheckpointRequest + 60, // 73: atelet.Atelet.Restore:input_type -> atelet.RestoreRequest + 62, // 74: atelet.Atelet.PromoteSnapshot:input_type -> atelet.PromoteSnapshotRequest + 64, // 75: atelet.Atelet.DetachActor:input_type -> atelet.DetachActorRequest + 31, // 76: atelet.Atelet.Terminate:input_type -> atelet.TerminateRequest + 30, // 77: atelet.WorkerSupport.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse + 5, // 78: atelet.WorkerSupport.AcquireWorkerEpoch:output_type -> atelet.AcquireWorkerEpochResponse + 17, // 79: atelet.WorkerSupport.RegisterWorker:output_type -> atelet.RegisterWorkerResponse + 8, // 80: atelet.WorkerSupport.SetWorkerSandboxRuntimes:output_type -> atelet.SetWorkerSandboxRuntimesResponse + 19, // 81: atelet.WorkerSupport.RequestActorSuspend:output_type -> atelet.RequestActorSuspendResponse + 23, // 82: atelet.WorkerSupport.GetActorEgressPolicy:output_type -> atelet.GetActorEgressPolicyResponse + 21, // 83: atelet.WorkerSupport.ReportActorCrashed:output_type -> atelet.ReportActorCrashedResponse + 52, // 84: atelet.Atelet.Run:output_type -> atelet.RunResponse + 59, // 85: atelet.Atelet.Checkpoint:output_type -> atelet.CheckpointResponse + 61, // 86: atelet.Atelet.Restore:output_type -> atelet.RestoreResponse + 63, // 87: atelet.Atelet.PromoteSnapshot:output_type -> atelet.PromoteSnapshotResponse + 65, // 88: atelet.Atelet.DetachActor:output_type -> atelet.DetachActorResponse + 32, // 89: atelet.Atelet.Terminate:output_type -> atelet.TerminateResponse + 77, // [77:90] is the sub-list for method output_type + 64, // [64:77] is the sub-list for method input_type + 64, // [64:64] is the sub-list for extension type_name + 64, // [64:64] is the sub-list for extension extendee + 0, // [0:64] is the sub-list for field type_name } func init() { file_atelet_proto_init() } diff --git a/internal/proto/ateletpb/atelet.proto b/internal/proto/ateletpb/atelet.proto index f68551c7b6..5ed0cee19d 100644 --- a/internal/proto/ateletpb/atelet.proto +++ b/internal/proto/ateletpb/atelet.proto @@ -480,7 +480,9 @@ service Atelet { rpc DetachActor(DetachActorRequest) returns (DetachActorResponse) {} // Terminate tells atelet to terminate/kill any running workload for an actor, - // unmount its volumes, and clean up actor state on the node. + // unmount its volumes, and clean up actor state on the node. It returns the + // actor's state as atelet forgets it, so the caller can record an upload + // atelet has not reported yet. rpc Terminate(TerminateRequest) returns (TerminateResponse) {} } @@ -522,6 +524,11 @@ message TerminateRequest { } message TerminateResponse { + // The actor's state as atelet forgets it: latest_local and uploading are + // unset, and latest_durable is its latest snapshot atelet knew was durable, + // if any. Terminate cancels an upload in progress, so that upload's snapshot + // is latest_durable only if the upload finished first. + ActorSnapshotState state = 1; } message RunRequest { diff --git a/internal/proto/ateletpb/atelet_grpc.pb.go b/internal/proto/ateletpb/atelet_grpc.pb.go index a3c2a54c82..328f9a8373 100644 --- a/internal/proto/ateletpb/atelet_grpc.pb.go +++ b/internal/proto/ateletpb/atelet_grpc.pb.go @@ -500,7 +500,9 @@ type AteletClient interface { // snapshot durable, such as when it has no `snapshot.object`. DetachActor(ctx context.Context, in *DetachActorRequest, opts ...grpc.CallOption) (*DetachActorResponse, error) // Terminate tells atelet to terminate/kill any running workload for an actor, - // unmount its volumes, and clean up actor state on the node. + // unmount its volumes, and clean up actor state on the node. It returns the + // actor's state as atelet forgets it, so the caller can record an upload + // atelet has not reported yet. Terminate(ctx context.Context, in *TerminateRequest, opts ...grpc.CallOption) (*TerminateResponse, error) } @@ -631,7 +633,9 @@ type AteletServer interface { // snapshot durable, such as when it has no `snapshot.object`. DetachActor(context.Context, *DetachActorRequest) (*DetachActorResponse, error) // Terminate tells atelet to terminate/kill any running workload for an actor, - // unmount its volumes, and clean up actor state on the node. + // unmount its volumes, and clean up actor state on the node. It returns the + // actor's state as atelet forgets it, so the caller can record an upload + // atelet has not reported yet. Terminate(context.Context, *TerminateRequest) (*TerminateResponse, error) mustEmbedUnimplementedAteletServer() } diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index e453a318c9..c7f643be7d 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -2729,7 +2729,8 @@ service WorkerService { // atelet detached the Actor from the node. Only atelet observes either. // // atelet calls this with its own client certificate, for an Actor on its - // node. Each report carries the whole state, so repeating one is safe. + // node. Each report carries the whole state, so repeating one is safe. For a + // CRASHED Actor, the control plane records only the finished upload. // Returns ABORTED while another operation is changing the Actor; retry. // Returns NOT_FOUND once the Actor is gone or no longer on the caller's // node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report diff --git a/pkg/proto/ateapipb/ateapi_grpc.pb.go b/pkg/proto/ateapipb/ateapi_grpc.pb.go index f8a5e5be72..429f0610e5 100644 --- a/pkg/proto/ateapipb/ateapi_grpc.pb.go +++ b/pkg/proto/ateapipb/ateapi_grpc.pb.go @@ -1988,7 +1988,8 @@ type WorkerServiceClient interface { // atelet detached the Actor from the node. Only atelet observes either. // // atelet calls this with its own client certificate, for an Actor on its - // node. Each report carries the whole state, so repeating one is safe. + // node. Each report carries the whole state, so repeating one is safe. For a + // CRASHED Actor, the control plane records only the finished upload. // Returns ABORTED while another operation is changing the Actor; retry. // Returns NOT_FOUND once the Actor is gone or no longer on the caller's // node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report @@ -2186,7 +2187,8 @@ type WorkerServiceServer interface { // atelet detached the Actor from the node. Only atelet observes either. // // atelet calls this with its own client certificate, for an Actor on its - // node. Each report carries the whole state, so repeating one is safe. + // node. Each report carries the whole state, so repeating one is safe. For a + // CRASHED Actor, the control plane records only the finished upload. // Returns ABORTED while another operation is changing the Actor; retry. // Returns NOT_FOUND once the Actor is gone or no longer on the caller's // node. Returns FAILED_PRECONDITION or INVALID_ARGUMENT when the report