Reclaim actor node state when a worker pod goes away

cmd/ateapi/internal/controlapi/service.go

@@ -102,7 +102,14 @@ func NewRPCService(
 		actorIDCAPool:          actorIDCAPool,
 	}
 	s.actorWorkflow = NewActorWorkflow(impl, workerCache, dialer, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s, objectStore)
-	s.workerWorkflow = NewWorkerWorkflow(impl)
+	// Converted through an explicit nil check: a nil *AteletDialer assigned
+	// straight to the interface would be a non-nil interface holding a nil
+	// pointer, and the worker workflow's nil check would not see it.
+	var nodeDialer ateletNodeDialer
+	if dialer != nil {
+		nodeDialer = dialer
+	}
+	s.workerWorkflow = NewWorkerWorkflow(impl, nodeDialer)
 	return s
 }
 

cmd/ateapi/internal/controlapi/worker_test.go

@@ -96,7 +96,7 @@ func newWorkerAPIService(t *testing.T) (*RPCService, store.Interface) {
 	persistence, cleanup := storetest.SetupTestStore(t)
 	t.Cleanup(cleanup)
 	impl := newServiceImpl(persistence, nil)
-	return &RPCService{impl: impl, workerWorkflow: NewWorkerWorkflow(persistence)}, persistence
+	return &RPCService{impl: impl, workerWorkflow: NewWorkerWorkflow(persistence, nil)}, persistence
 }
 
 // seedAPIWorker registers a worker directly through the store and returns it as

cmd/ateapi/internal/controlapi/workflow.go

@@ -32,6 +32,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc"
 	grpcCodes "google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	storagev1listers "k8s.io/client-go/listers/storage/v1"
@@ -170,12 +171,22 @@ type actorWorkflowStore interface {
 // does from the other side: releasing the Actor bound to a Worker stays
 // in-process because there is no bind/release RPC.
 type WorkerWorkflow struct {
-	store workerWorkflowStore
+	store  workerWorkflowStore
+	dialer ateletNodeDialer
 }
 
-// NewWorkerWorkflow creates a new WorkerWorkflow.
-func NewWorkerWorkflow(store workerWorkflowStore) *WorkerWorkflow {
-	return &WorkerWorkflow{store: store}
+// ateletNodeDialer reaches the atelet on a named node. Narrower than the
+// *AteletDialer the actor workflows hold, because a Worker being deleted has
+// lost its pod: the by-pod lookup those use cannot resolve, and the node is
+// the only handle left on the atelet still holding the actor's state.
+type ateletNodeDialer interface {
+	DialForAteletOnNode(nodeName string) (*grpc.ClientConn, error)
+}
+
+// NewWorkerWorkflow creates a new WorkerWorkflow. A nil dialer disables the
+// node-state reclaim step (tests that assert only on store transitions).
+func NewWorkerWorkflow(store workerWorkflowStore, dialer ateletNodeDialer) *WorkerWorkflow {
+	return &WorkerWorkflow{store: store, dialer: dialer}
 }
 
 // workerWorkflowStore enumerates the exact storage methods needed by

cmd/ateapi/internal/controlapi/workflow_delete.go

@@ -176,23 +176,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 			slog.String("actor", actorRef.Name),
 			slog.String("templateAtespace", actor.GetActorTemplate().GetAtespace()),
 			slog.String("templateName", actor.GetActorTemplate().GetName()))
-		workloadSpec = &ateletpb.WorkloadSpec{}
-		for _, vol := range actor.GetStatus().GetActorVolumes() {
-			// StorageVolumeId is only populated once the volume is provisioned.
-			// Skip volumes that were never created (e.g. failed during PENDING state).
-			if vol.GetStorageVolumeId() != "" {
-				workloadSpec.Volumes = append(workloadSpec.Volumes, &ateletpb.Volume{
-					Name: vol.GetVolumeName(),
-					Source: &ateletpb.Volume_External{
-						External: &ateletpb.ExternalVolumeSource{
-							StorageVolumeId: vol.GetStorageVolumeId(),
-							VolumeType:      vol.GetVolumeType(),
-							VolumeContext:   vol.GetVolumeContext(),
-						},
-					},
-				})
-			}
-		}
+		workloadSpec = fallbackWorkloadSpec(actor)
 	}
 
 	req := &ateletpb.TerminateRequest{
@@ -216,6 +200,32 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 	return nil
 }
 
+// fallbackWorkloadSpec is the spec to terminate an actor with when its template
+// cannot be resolved — deleted, or never readable from the caller in the first
+// place. Only the external volumes recorded on the actor are carried, because
+// they are the one part of the spec a teardown still acts on: atelet unmounts
+// them from the node. A volume with no StorageVolumeId was never provisioned
+// (e.g. the actor failed while PENDING), so there is nothing mounted to name.
+func fallbackWorkloadSpec(actor *ateapipb.Actor) *ateletpb.WorkloadSpec {
+	spec := &ateletpb.WorkloadSpec{}
+	for _, vol := range actor.GetStatus().GetActorVolumes() {
+		if vol.GetStorageVolumeId() == "" {
+			continue
+		}
+		spec.Volumes = append(spec.Volumes, &ateletpb.Volume{
+			Name: vol.GetVolumeName(),
+			Source: &ateletpb.Volume_External{
+				External: &ateletpb.ExternalVolumeSource{
+					StorageVolumeId: vol.GetStorageVolumeId(),
+					VolumeType:      vol.GetVolumeType(),
+					VolumeContext:   vol.GetVolumeContext(),
+				},
+			},
+		})
+	}
+	return spec
+}
+
 // ensureVolumesDetachedForDelete detaches external volumes.
 func (w *ActorWorkflow) ensureVolumesDetachedForDelete(ctx context.Context, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (err error) {
 	ctx, done := stepSpan(ctx, "DetachVolumesForDelete")

cmd/ateapi/internal/controlapi/workflow_delete_test.go

@@ -443,7 +443,7 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) {
 
 			// The worker's pod goes away with the commit still outstanding, so
 			// the suspend never gets to finish.
-			if _, err := NewWorkerWorkflow(persistence).DeleteWorker(ctx, workerName, store.DeletePreconditions{}); err != nil {
+			if _, err := NewWorkerWorkflow(persistence, nil).DeleteWorker(ctx, workerName, store.DeletePreconditions{}); err != nil {
 				t.Fatalf("DeleteWorker: %v", err)
 			}
 

cmd/ateapi/internal/controlapi/workflow_worker_delete.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/internal/ateattr"
+	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
 	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"google.golang.org/grpc/codes"
@@ -187,6 +188,14 @@ func (w *WorkerWorkflow) releaseBoundActor(ctx context.Context, worker *ateapipb
 		markSkipped(ctx, "actor no longer points at this worker")
 		return nil
 	}
+	// Reclaim before the state checks below: the disk the actor left on the
+	// node has to go whether it suspended cleanly or crashed, and this is the
+	// last moment anything still knows which node that is. Once the record is
+	// released, the actor names no worker, no worker names a node, and nothing
+	// ever revisits the actor's UID — the directories are orphaned for the life
+	// of the node.
+	w.reclaimActorStateOnNode(ctx, worker, actor)
+
 	// If the actor is suspended, it's already been released.
 	if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED {
 		markSkipped(ctx, "actor suspended cleanly before the pod went away")
@@ -238,6 +247,87 @@ func (w *WorkerWorkflow) releaseBoundActor(ctx context.Context, worker *ateapipb
 	return nil
 }
 
+// reclaimActorStateOnNode asks the atelet on the worker's node to terminate the
+// actor, which is what reclaims the actor's state directory
+// (/var/lib/ateom-gvisor/actors/<uid>: durable dir, checkpoint and restore
+// images — gigabytes for a durdir actor) and unmounts its external volumes.
+//
+// The worker's pod is gone by the time this runs, so the atelet is reached by
+// node: the by-pod lookup the actor workflows use resolves through the pod that
+// has just disappeared, and returning "pod not found" there is exactly how this
+// state came to be orphaned. atelet tolerates the ateom being gone and reclaims
+// the directories anyway.
+//
+// Best-effort by construction: a node that cannot be reached must not wedge the
+// deregistration of its workers, and the reclaim has no record of its own to
+// retry from once the worker is deleted. What it misses — an unreachable atelet,
+// a node that never comes back, an abrupt eviction — is the orphan sweep's to
+// collect.
+func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *ateapipb.Worker, actor *ateapipb.Actor) {
+	ctx, done := stepSpan(ctx, "ReclaimActorStateOnNode")
+	defer func() { _ = done(nil) }()
+
+	if w.dialer == nil {
+		markSkipped(ctx, "no atelet dialer configured")
+		return
+	}
+	nodeName := worker.GetNodeName()
+	if nodeName == "" {
+		// Pre-dates the field, or a Worker registered before its pod was
+		// scheduled. Nothing names the node holding the state.
+		markSkipped(ctx, "worker records no node")
+		return
+	}
+	// A local snapshot is state this node holds deliberately, and Terminate
+	// prunes local checkpoints. Leaving it is the conservative choice: a
+	// wrongly-kept snapshot costs disk the sweep can still reclaim later, a
+	// wrongly-deleted one cannot be recovered at all.
+	if actor.GetStatus().GetLocalSnapshotInfo() != nil {
+		markSkipped(ctx, "actor has a local snapshot pinned to the node")
+		return
+	}
+
+	actorRef := resources.ActorRefFromActor(actor)
+	logAttrs := []any{
+		slog.Any("actor", actorRef),
+		slog.String("actor_uid", actor.GetMetadata().GetUid()),
+		slog.String("worker", worker.GetMetadata().GetName()),
+		slog.String("node", nodeName),
+	}
+
+	conn, err := w.dialer.DialForAteletOnNode(nodeName)
+	if err != nil {
+		// Includes ErrNoAteletOnNode: the atelet is restarting, or the node
+		// itself is gone. Nothing to reclaim against right now.
+		slog.WarnContext(ctx, "Could not reach the atelet holding a released actor's state; leaving it for the orphan sweep",
+			append(logAttrs, slog.Any("err", err))...)
+		return
+	}
+
+	slog.InfoContext(ctx, "Reclaiming the node state of an actor released from a worker whose pod is gone", logAttrs...)
+	// The template is not resolvable from this workflow — and would be the
+	// wrong thing to block on if it were, since a delete can outlive it. The
+	// fallback spec carries what the teardown acts on: the external volumes to
+	// unmount.
+	_, err = ateletpb.NewAteomHerderClient(conn).Terminate(ctx, &ateletpb.TerminateRequest{
+		TargetAteomUid:        worker.GetWorkerPodUid(),
+		Atespace:              actor.GetMetadata().GetAtespace(),
+		ActorName:             actor.GetMetadata().GetName(),
+		ActorUid:              actor.GetMetadata().GetUid(),
+		ActorTemplateAtespace: actor.GetActorTemplate().GetAtespace(),
+		ActorTemplateName:     actor.GetActorTemplate().GetName(),
+		Spec:                  fallbackWorkloadSpec(actor),
+	})
+	switch {
+	case err == nil:
+	case status.Code(err) == codes.NotFound:
+		slog.InfoContext(ctx, "Actor already terminated on its node", logAttrs...)
+	default:
+		slog.WarnContext(ctx, "Failed to reclaim a released actor's node state; leaving it for the orphan sweep",
+			append(logAttrs, slog.Any("err", err))...)
+	}
+}
+
 // finalizeDeleted removes the worker from the store and returns the deleted
 // record. The request's guards are carried down as delete preconditions, so a
 // worker that moved on since the caller read it is reported as a conflict rather