Make checkpoint and delete survive control-plane and object-store disruption during upgrades

cmd/ateapi/internal/controlapi/dialer.go

@@ -35,6 +35,8 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/lru"
 )
@@ -44,6 +46,12 @@ import (
 // Retryable.
 var ErrNoAteletOnNode = errors.New("no atelet pod found on node")
 
+// ErrNodeGone reports that the requested node no longer exists in the
+// cluster, so no atelet will ever serve on it again: whatever an actor left
+// there is gone with the node. Not retryable. Reported only by a dialer built
+// WithNodeLister; without one a gone node reports ErrNoAteletOnNode.
+var ErrNodeGone = errors.New("node no longer exists")
+
 // ateletConn is a cached connection and the pod IP it was dialed on.
 type ateletConn struct {
 	ip   string
@@ -53,6 +61,9 @@ type ateletConn struct {
 // AteletDialer handles gRPC connections to Atelet pods.
 type AteletDialer struct {
 	ateletIndexer cache.Indexer
+	// nodeLister, when set, tells a node that is gone (ErrNodeGone) from one
+	// whose atelet is merely absent (ErrNoAteletOnNode).
+	nodeLister corev1listers.NodeLister
 	// mu makes the lookup, stale eviction and insert in DialForAteletOnNode
 	// atomic, so concurrent callers cannot evict each other's fresh conn.
 	mu sync.Mutex
@@ -74,6 +85,12 @@ func WithDialCredentials(build func(expectedPodUID string) (credentials.Transpor
 	return func(d *AteletDialer) { d.dialCredentials = build }
 }
 
+// WithNodeLister lets the dialer report ErrNodeGone for a node the cluster no
+// longer has, instead of ErrNoAteletOnNode.
+func WithNodeLister(nodes corev1listers.NodeLister) DialerOption {
+	return func(d *AteletDialer) { d.nodeLister = nodes }
+}
+
 // NewAteletDialer creates a new AteletDialer. clientBundlePath and serverCAPath
 // are used to build the per-atelet mTLS credentials used for every atelet
 // connection, and ateletSPIFFEID is the identity those credentials expect on
@@ -125,6 +142,11 @@ func (d *AteletDialer) DialForAteletOnNode(nodeName string) (*grpc.ClientConn, e
 	}
 
 	if len(matchingAtelets) == 0 {
+		if d.nodeLister != nil {
+			if _, err := d.nodeLister.Get(nodeName); apierrors.IsNotFound(err) {
+				return nil, fmt.Errorf("%w: %q", ErrNodeGone, nodeName)
+			}
+		}
 		return nil, fmt.Errorf("%w: %q", ErrNoAteletOnNode, nodeName)
 	}
 	if len(matchingAtelets) > 1 {

cmd/ateapi/internal/controlapi/dialer_test.go

@@ -41,6 +41,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 )
 
@@ -330,6 +331,21 @@ func TestDialForAteletOnNode(t *testing.T) {
 		}
 	})
 
+	t.Run("no atelet on a node the cluster lost", func(t *testing.T) {
+		nodes := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+		if err := nodes.Add(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node2"}}); err != nil {
+			t.Fatalf("adding node to indexer: %v", err)
+		}
+		d := NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "",
+			WithNodeLister(corev1listers.NewNodeLister(nodes)))
+		if _, err := d.DialForAteletOnNode("node1"); !errors.Is(err, ErrNodeGone) {
+			t.Fatalf("DialForAteletOnNode(gone node) = %v, want ErrNodeGone", err)
+		}
+		if _, err := d.DialForAteletOnNode("node2"); !errors.Is(err, ErrNoAteletOnNode) {
+			t.Fatalf("DialForAteletOnNode(node without atelet) = %v, want ErrNoAteletOnNode", err)
+		}
+	})
+
 	t.Run("more than one atelet on node", func(t *testing.T) {
 		d := NewAteletDialer(newTestAteletIndexer(t,
 			ateletPod("atelet-1", "uid-1", "node1", "10.0.0.1"),

cmd/ateapi/internal/controlapi/workflow_delete.go

@@ -155,6 +155,16 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 
 	conn, err := w.dialer.DialForAteletOnNode(nodeName)
 	if err != nil {
+		// The node went away with everything the actor left on it: its
+		// workload, local snapshots and mounts. There is nothing to terminate,
+		// and retrying would hold the actor in DELETING for good (the reaper
+		// re-deletes a DELETING actor every pass).
+		if errors.Is(err, ErrNodeGone) {
+			slog.InfoContext(ctx, "node that hosted the actor no longer exists; skipping atelet terminate request",
+				slog.String("node", nodeName),
+				slog.Any("actor", actorRef))
+			return nil
+		}
 		return fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err)
 	}
 

cmd/ateapi/internal/controlapi/workflow_delete_test.go

@@ -23,12 +23,15 @@ import (
 	"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/installdefaults"
 	"go.ate.dev/substrate/internal/objectstore/objectstoretest"
 	"go.ate.dev/substrate/internal/objectstoreplugin/objectstoreplugintest"
 	"go.ate.dev/substrate/internal/resources"
 	"go.ate.dev/substrate/pkg/proto/ateapipb"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
 )
 
 func TestDeleteActorWorkflow_ExecutionPaths(t *testing.T) {
@@ -511,3 +514,39 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) {
 		})
 	}
 }
+
+// A crashed actor keeps the node it ran on as a record. When that node has
+// left the cluster, deleting the actor has nothing to terminate there and must
+// not wait for an atelet that never comes back.
+func TestDeleteActor_SkipsTerminateOnGoneNode(t *testing.T) {
+	ctx := context.Background()
+	st, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+	w := newTestActorWorkflow(t, st, "ns", "tmpl1")
+
+	actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"}
+	seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_CRASHED, func(a *ateapipb.Actor) {
+		a.Status.AssignedNode = "gone-node"
+	})
+
+	// The node is absent from the lister and no atelet is indexed on it.
+	noNodes := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+	w.dialer = NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "",
+		WithNodeLister(corev1listers.NewNodeLister(noNodes)))
+
+	if _, err := w.DeleteActor(ctx, actorRef, false, store.DeletePreconditions{}); err != nil {
+		t.Fatalf("DeleteActor on a gone node: %v", err)
+	}
+	if _, err := st.GetActor(ctx, actorRef); !errors.Is(err, store.ErrNotFound) {
+		t.Fatalf("GetActor after delete: err = %v, want ErrNotFound", err)
+	}
+
+	// Without a node lister the same delete keeps retrying, as before.
+	seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_CRASHED, func(a *ateapipb.Actor) {
+		a.Status.AssignedNode = "gone-node"
+	})
+	w.dialer = newDanglingDialer()
+	if _, err := w.DeleteActor(ctx, actorRef, false, store.DeletePreconditions{}); !errors.Is(err, ErrNoAteletOnNode) {
+		t.Fatalf("DeleteActor without a node lister: err = %v, want ErrNoAteletOnNode", err)
+	}
+}

cmd/ateapi/main.go

@@ -284,6 +284,9 @@ func main() {
 	ateletPodInformerFactory, ateletPodInformer := controlapi.AteletInformer(clientset, ateletNamespace)
 	scInformerFactory := informers.NewSharedInformerFactory(clientset, 0)
 	storageClassLister := scInformerFactory.Storage().V1().StorageClasses().Lister()
+	// Lets the atelet dialer tell a node the cluster lost from one whose
+	// atelet is restarting, so a delete does not wait forever on the former.
+	nodeLister := scInformerFactory.Core().V1().Nodes().Lister()
 
 	stopCh := make(chan struct{})
 	defer close(stopCh)
@@ -322,7 +325,7 @@ func main() {
 	}
 
 	volPlugins := make(map[string]volume.VolumePluginControlPlane)
-	ateletDialer := controlapi.NewAteletDialer(ateletPodInformer.GetIndexer(), ateletSPIFFEID, *ateletClientCredBundle, *podIdentityCACerts)
+	ateletDialer := controlapi.NewAteletDialer(ateletPodInformer.GetIndexer(), ateletSPIFFEID, *ateletClientCredBundle, *podIdentityCACerts, controlapi.WithNodeLister(nodeLister))
 
 	actorIDCAPool, err := localca.NewRefreshingPool(*actorIDCAPoolFile)
 	if err != nil {

cmd/atelet/internal/snapshottracker/lifecycle.go

@@ -215,6 +215,37 @@ func (t *Tracker) BeginCheckpoint(ctx context.Context, actorUID string, snap *at
 	return t.holdLocked(actorUID, a, &SnapshotRef{Snapshot: proto.CloneOf(snap)}, false), nil
 }
 
+// BeginUploadingCheckpoint is BeginCheckpoint for a Checkpoint under
+// UPLOAD_MODE_IMMEDIATE, which uploads snap itself: it also waits for a
+// background upload of snap in progress, and holds the actor's background
+// uploads off until the checkpoint ends, as BeginPromote does, so the same
+// snapshot is never uploaded twice at once.
+func (t *Tracker) BeginUploadingCheckpoint(ctx context.Context, actorUID string, snap *ateletpb.Snapshot) (end func(), err error) {
+	t.lock(actorUID)
+	defer t.unlock(actorUID)
+	a, err := t.waitForDetachOrSnapshotUploadLocked(ctx, actorUID, SnapshotRef{Snapshot: snap})
+	if err != nil {
+		return nil, err
+	}
+	return t.holdLocked(actorUID, a, &SnapshotRef{Snapshot: proto.CloneOf(snap)}, true), nil
+}
+
+// Committed reports whether snap is already recorded as the actor's latest
+// local snapshot, and whether as its latest durable one. A retried Checkpoint
+// of a snapshot an earlier attempt committed, whose response was lost or
+// whose upload failed, finds one or both set and must not drive worker again:
+// the workload is gone.
+func (t *Tracker) Committed(actorUID string, snap *ateletpb.Snapshot) (local, durable bool) {
+	t.lock(actorUID)
+	defer t.unlock(actorUID)
+	a := t.findLocked(actorUID)
+	if a == nil {
+		return false, false
+	}
+	ref := SnapshotRef{Snapshot: snap}
+	return sameUUID(a.LatestLocal, &ref), sameUUID(a.LatestDurable, &ref)
+}
+
 // BeginPromote records that a PromoteSnapshot of snap is in progress on this
 // node, once any detach in progress and any background upload of snap have
 // finished. It fails with FailedPrecondition when snap has a stale generation

cmd/atelet/lifecycle_test.go

@@ -1267,51 +1267,107 @@ func TestImmediateCheckpointWithoutLocalCopy(t *testing.T) {
 // recorded as the actor's latest local snapshot, so the actor can still be
 // restored from it.
 func TestFailedCheckpointKeepsLocalSnapshot(t *testing.T) {
-	for _, tc := range []struct {
-		name string
-		snap string
-		mode ateletpb.UploadMode
-		fail func(t *testing.T, a *pluginActor)
-	}{
-		{
-			name: "suspend upload fails",
-			snap: "suspend-snap-2",
-			mode: ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE,
-			fail: func(t *testing.T, a *pluginActor) {
-				a.store.putErr = errors.New("object store unavailable")
-			},
-		},
-		{
-			name: "pause move fails",
-			snap: "pause-snap-2",
-			mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED,
-			fail: func(t *testing.T, a *pluginActor) {
-				// A file where the new snapshot's directory belongs fails the move.
-				if err := os.WriteFile(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-2"), nil, 0o600); err != nil {
-					t.Fatal(err)
-				}
-			},
-		},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			a := runPluginActor(t)
-			if _, err := a.checkpoint(t, a.snapshot("pause-snap-1", 1), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil {
-				t.Fatalf("Checkpoint (pause): %v", err)
-			}
-			a.restoreLocal(t, "pause-snap-1", 1)
+	a := runPluginActor(t)
+	if _, err := a.checkpoint(t, a.snapshot("pause-snap-1", 1), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil {
+		t.Fatalf("Checkpoint (pause): %v", err)
+	}
+	a.restoreLocal(t, "pause-snap-1", 1)
 
-			tc.fail(t, a)
-			if _, err := a.checkpoint(t, a.snapshot(tc.snap, 2), tc.mode); err == nil {
-				t.Fatal("Checkpoint succeeded, want it to fail")
-			}
-			if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-1")); err != nil {
-				t.Errorf("pause snapshot after the failed checkpoint: stat = %v, want it kept", err)
-			}
-			if got := a.s.snapshots.State(pluginActorUID).GetLatestLocal().GetSnapshot().GetSnapshotUuid(); got != "pause-snap-1" {
-				t.Errorf("latest_local = %q, want pause-snap-1", got)
-			}
-			a.restoreLocal(t, "pause-snap-1", 1)
-		})
+	// A file where the new snapshot's directory belongs fails the move.
+	if err := os.WriteFile(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-2"), nil, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := a.checkpoint(t, a.snapshot("pause-snap-2", 2), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err == nil {
+		t.Fatal("Checkpoint succeeded, want it to fail")
+	}
+	if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-1")); err != nil {
+		t.Errorf("pause snapshot after the failed checkpoint: stat = %v, want it kept", err)
+	}
+	if got := a.s.snapshots.State(pluginActorUID).GetLatestLocal().GetSnapshot().GetSnapshotUuid(); got != "pause-snap-1" {
+		t.Errorf("latest_local = %q, want pause-snap-1", got)
+	}
+	a.restoreLocal(t, "pause-snap-1", 1)
+}
+
+// transientStoreError is a storage service failure that reports a transient
+// HTTP status, as a backend SDK's response error does once its retries are
+// exhausted.
+type transientStoreError struct{ code int }
+
+func (e *transientStoreError) Error() string {
+	return fmt.Sprintf("storage service answered %d", e.code)
+}
+func (e *transientStoreError) HTTPStatusCode() int { return e.code }
+
+// TestSuspendUploadFailureRetriesFromLocalCopy suspends a resumed actor while
+// the object store is failing over: the checkpoint reports the outage as
+// Unavailable, so the control plane retries rather than crashes the actor, and
+// keeps the snapshot worker wrote on the node, recorded as the actor's latest
+// local snapshot. The retried Checkpoint uploads that copy without driving
+// worker again (its workload is gone), and records it durable with the
+// checkpoint's commit time. A further retry reports the same state without
+// uploading again.
+func TestSuspendUploadFailureRetriesFromLocalCopy(t *testing.T) {
+	a := runPluginActor(t)
+	if _, err := a.checkpoint(t, a.snapshot("pause-snap-1", 1), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil {
+		t.Fatalf("Checkpoint (pause): %v", err)
+	}
+	a.restoreLocal(t, "pause-snap-1", 1)
+	snap := a.snapshot("suspend-snap-2", 2)
+
+	a.store.putErr = &transientStoreError{503}
+	_, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE)
+	if apierror.Code(err) != codes.Unavailable {
+		t.Fatalf("Checkpoint during the outage = %v, want %s", err, codes.Unavailable)
+	}
+	if a.worker.actorDirs["CheckpointWorkload"] == nil {
+		t.Fatal("worker.CheckpointWorkload not called by the first Checkpoint")
+	}
+	if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "suspend-snap-2")); err != nil {
+		t.Errorf("snapshot after the failed upload: stat = %v, want it kept", err)
+	}
+	failed := a.s.snapshots.State(pluginActorUID)
+	if got := failed.GetLatestLocal(); !proto.Equal(got.GetSnapshot(), snap) || got.GetLocalCommitTime() == nil {
+		t.Errorf("latest_local after the failed upload = %v, want %v with its commit time", got, snap)
+	}
+	if failed.GetLatestDurable() != nil {
+		t.Errorf("latest_durable after the failed upload = %v, want unset", failed.GetLatestDurable())
+	}
+
+	a.store.putErr = nil
+	delete(a.worker.actorDirs, "CheckpointWorkload")
+	resp, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE)
+	if err != nil {
+		t.Fatalf("retried Checkpoint: %v", err)
+	}
+	if got := a.worker.actorDirs["CheckpointWorkload"]; got != nil {
+		t.Errorf("retried Checkpoint called worker.CheckpointWorkload (%v), want the local copy uploaded instead", got)
+	}
+	if !slices.Contains(a.store.keys(), manifestKey("suspend-snap-2")) {
+		t.Errorf("objects = %v, want the snapshot uploaded", a.store.keys())
+	}
+	if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "suspend-snap-2")); err != nil {
+		t.Errorf("local snapshot after the retried checkpoint: stat = %v, want it kept", err)
+	}
+	st := resp.GetState()
+	commitTime := failed.GetLatestLocal().GetLocalCommitTime()
+	if got := st.GetLatestLocal(); !proto.Equal(got.GetSnapshot(), snap) || !proto.Equal(got.GetLocalCommitTime(), commitTime) {
+		t.Errorf("latest_local = %v, want %v committed at %v", got, snap, commitTime)
+	}
+	if got := st.GetLatestDurable(); !proto.Equal(got.GetSnapshot(), snap) || !proto.Equal(got.GetLocalCommitTime(), commitTime) || got.GetDurableCompleteTime() == nil {
+		t.Errorf("latest_durable = %v, want %v committed at %v, with its upload time", got, snap, commitTime)
+	}
+	if got := a.s.snapshots.State(pluginActorUID); !proto.Equal(got, st) {
+		t.Errorf("tracked state = %v, want %v, as Checkpoint returned", got, st)
+	}
+
+	a.store.putErr = errors.New("retried suspend should not upload again")
+	again, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE)
+	if err != nil {
+		t.Fatalf("Checkpoint retried after the upload: %v", err)
+	}
+	if got := again.GetState(); !proto.Equal(got, st) {
+		t.Errorf("state after a second retry = %v, want %v", got, st)
 	}
 }