Decouple actor lock TTL from workflow deadline via heartbeat (#14)

ActorWorkflow.ResumeActor and SuspendActor used to derive their workflow
ctx from the Redis lock TTL via acquireActorLock(ctx, id, 30s, 2s) — the
workflow deadline and the lock TTL were a single 28s knob. That meant
image pulls / restores that legitimately need more than 28s death-looped
forever, while raising the knob also raised how long peers wait to retry
an actor after a crashed ateapi replica.

Split the two concerns:

- Lock TTL stays short (30s constant, internal). Bounds peer failover.
- Workflow deadline is a separate operator-configurable knob via the
  new --actor-workflow-deadline pflag (default 5m). Bounds a single
  Resume/Suspend.
- A heartbeat goroutine refreshes the lock every lockTTL/3 (~10s) for
  the full workflow duration. On RefreshLock=false or any Redis error
  (peer stole the lock, Redis blip), the workflow ctx is cancelled with
  errLostActorLock as the cause so in-flight steps unwind cleanly and
  the mutual-exclusion invariant is preserved.
- The release function stops the heartbeat (waits for goroutine exit)
  before best-effort ReleaseLock.

Adds store.Interface.RefreshLock with a Redis CAS Lua script mirroring
the existing ReleaseLock script.

Signed-off-by: Eitan Yarmush <eitan.yarmush@solo.io>

Author: jjamroga

charts/substrate-crds/templates/ate.dev_actortemplates.yaml

@@ -226,6 +226,74 @@ spec:
                   type: object
                 maxItems: 10
                 type: array
+              resources:
+                description: |-
+                  Resources declares the compute resources for each actor of this template.
+                  Unlike a pod, an actor is sized by its Limits: the sandbox is built to the
+                  CPU/memory limits (cgroup caps, and for the micro-VM the VM's vCPU count and
+                  memory), the scheduler only places the actor on a worker whose capacity is
+                  >= these limits, and the limits are supplied to the sandbox over the actor
+                  RPCs. Because the size is baked into snapshots, it is part of the immutable
+                  spec. Requests and claims are not supported (actors are sized by limits only).
+                  A zero or absent limit leaves the sandbox at the runtime default (unlimited
+                  for gVisor, the kata config for the micro-VM).
+                properties:
+                  claims:
+                    description: |-
+                      Claims lists the names of resources, defined in spec.resourceClaims,
+                      that are used by this container.
+
+                      This field depends on the
+                      DynamicResourceAllocation feature gate.
+
+                      This field is immutable. It can only be set for containers.
+                    items:
+                      description: ResourceClaim references one entry in PodSpec.ResourceClaims.
+                      properties:
+                        name:
+                          description: |-
+                            Name must match the name of one entry in pod.spec.resourceClaims of
+                            the Pod where this field is used. It makes that resource available
+                            inside a container.
+                          type: string
+                        request:
+                          description: |-
+                            Request is the name chosen for a request in the referenced claim.
+                            If empty, everything from the claim is made available, otherwise
+                            only the result of this request.
+                          type: string
+                      required:
+                      - name
+                      type: object
+                    type: array
+                    x-kubernetes-list-map-keys:
+                    - name
+                    x-kubernetes-list-type: map
+                  limits:
+                    additionalProperties:
+                      anyOf:
+                      - type: integer
+                      - type: string
+                      pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                      x-kubernetes-int-or-string: true
+                    description: |-
+                      Limits describes the maximum amount of compute resources allowed.
+                      More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/
+                    type: object
+                  requests:
+                    additionalProperties:
+                      anyOf:
+                      - type: integer
+                      - type: string
+                      pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                      x-kubernetes-int-or-string: true
+                    description: |-
+                      Requests describes the minimum amount of compute resources required.
+                      If Requests is omitted for a container, it defaults to Limits if that is explicitly specified,
+                      otherwise to an implementation-defined value. Requests cannot exceed Limits.
+                      More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/
+                    type: object
+                type: object
               sandboxClass:
                 default: gvisor
                 description: |-
@@ -423,6 +491,17 @@ spec:
               rule: '(has(self.sandboxClass) && self.sandboxClass == ''microvm'')
                 || !has(self.snapshotsConfig.onResume) || (has(self.snapshotsConfig.onResume.fromData)
                 ? self.snapshotsConfig.onResume.fromData : ''ColdBoot'') != ''Golden'''
+            - message: spec.resources.requests is not supported; actors are sized
+                by spec.resources.limits only
+              rule: '!has(self.resources) || !has(self.resources.requests)'
+            - message: spec.resources.claims is not supported
+              rule: '!has(self.resources) || !has(self.resources.claims)'
+            - message: For sandboxClass 'microvm', spec.resources.limits.memory must
+                be at least 256Mi (128Mi VMM reserve + 128Mi guest minimum); below
+                this the VM cannot boot
+              rule: '!has(self.sandboxClass) || self.sandboxClass != ''microvm'' ||
+                !has(self.resources) || !has(self.resources.limits) || !(''memory''
+                in self.resources.limits) || !quantity(self.resources.limits[''memory'']).isLessThan(quantity(''256Mi''))'
           status:
             description: status is the observed state of ActorTemplate
             properties:

cmd/ateapi/internal/actoridentity/actoridentity.go

@@ -30,7 +30,6 @@ import (
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/actoridjwt"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
-	"github.com/agent-substrate/substrate/internal/k8sjwt"
 	"github.com/agent-substrate/substrate/internal/localca"
 	"github.com/agent-substrate/substrate/internal/localjwtauthority"
 	"github.com/agent-substrate/substrate/internal/principal"

cmd/ateapi/internal/controlapi/dialer_test.go

@@ -27,6 +27,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/agent-substrate/substrate/internal/installdefaults"
 	"github.com/agent-substrate/substrate/internal/substratex509"
 	"github.com/spiffe/go-spiffe/v2/bundle/x509bundle"
 	"github.com/spiffe/go-spiffe/v2/spiffeid"
@@ -214,7 +215,7 @@ func TestDialForWorkerTarget(t *testing.T) {
 				Spec:       corev1.PodSpec{NodeName: "node-1"},
 			}
 			ateletPod := &corev1.Pod{
-				ObjectMeta: metav1.ObjectMeta{Namespace: ateletNamespace, Name: "atelet-abc", UID: "atelet-uid"},
+				ObjectMeta: metav1.ObjectMeta{Namespace: installdefaults.SystemNamespace, Name: "atelet-abc", UID: "atelet-uid"},
 				Spec:       corev1.PodSpec{NodeName: "node-1"},
 				Status:     corev1.PodStatus{PodIPs: []corev1.PodIP{{IP: tc.ateletIP}}},
 			}
@@ -241,7 +242,7 @@ func TestDialForWorkerErrors(t *testing.T) {
 
 	t.Run("unknown worker pod", func(t *testing.T) {
 		ateletPod := &corev1.Pod{
-			ObjectMeta: metav1.ObjectMeta{Namespace: ateletNamespace, Name: "atelet-abc", UID: "atelet-uid"},
+			ObjectMeta: metav1.ObjectMeta{Namespace: installdefaults.SystemNamespace, Name: "atelet-abc", UID: "atelet-uid"},
 			Spec:       corev1.PodSpec{NodeName: "node-1"},
 			Status:     corev1.PodStatus{PodIPs: []corev1.PodIP{{IP: "10.244.1.7"}}},
 		}
@@ -253,7 +254,7 @@ func TestDialForWorkerErrors(t *testing.T) {
 
 	t.Run("atelet without assigned IPs", func(t *testing.T) {
 		ateletPod := &corev1.Pod{
-			ObjectMeta: metav1.ObjectMeta{Namespace: ateletNamespace, Name: "atelet-abc", UID: "atelet-uid"},
+			ObjectMeta: metav1.ObjectMeta{Namespace: installdefaults.SystemNamespace, Name: "atelet-abc", UID: "atelet-uid"},
 			Spec:       corev1.PodSpec{NodeName: "node-1"},
 		}
 		d := newDialerForPods(t, workerPod, ateletPod)

cmd/ateapi/internal/controlapi/functionaltest/common_test.go

@@ -61,10 +61,8 @@ const (
 	testAtespace = "test-atespace"
 	testActorID  = "id1"
 
-	// ateletNamespace and byNode mirror the unexported constants controlapi's
-	// atelet informer is built with.
-	ateletNamespace = "ate-system"
-	byNode          = "by-node"
+	// byNode mirrors the unexported index name controlapi's atelet informer uses.
+	byNode = "by-node"
 )
 
 var (
@@ -187,7 +185,7 @@ func setupTestWithVolumePlugins(t *testing.T, ns string, plugins map[string]volu
 			mockDriverName: mockPlugin,
 		}
 	}
-	service := controlapi.NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins)
+	service := controlapi.NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", 30*time.Second, volPlugins)
 
 	// 5. Start REAL gRPC Server for ATE API
 	grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ateinterceptors.ServerUnaryInterceptor))
@@ -594,15 +592,15 @@ func createAteletPod(kc kubernetes.Interface, name, nodeName string) error {
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
-			Namespace: ateletNamespace,
+			Namespace: installdefaults.SystemNamespace,
 			Labels:    map[string]string{"app": "atelet"},
 		},
 		Spec: corev1.PodSpec{
 			NodeName:   nodeName,
 			Containers: []corev1.Container{{Name: "main", Image: "nginx"}},
 		},
 	}
-	created, err := kc.CoreV1().Pods(ateletNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
+	created, err := kc.CoreV1().Pods(installdefaults.SystemNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
 	if apierrors.IsAlreadyExists(err) {
 		return nil
 	}
@@ -611,7 +609,7 @@ func createAteletPod(kc kubernetes.Interface, name, nodeName string) error {
 	}
 	created.Status.PodIPs = []corev1.PodIP{{IP: "127.0.0.1"}}
 	created.Status.Phase = corev1.PodRunning
-	if _, err := kc.CoreV1().Pods(ateletNamespace).UpdateStatus(context.Background(), created, metav1.UpdateOptions{}); err != nil {
+	if _, err := kc.CoreV1().Pods(installdefaults.SystemNamespace).UpdateStatus(context.Background(), created, metav1.UpdateOptions{}); err != nil {
 		return fmt.Errorf("updating atelet pod %s status: %w", name, err)
 	}
 	return nil
@@ -628,7 +626,7 @@ func setupAteletOnNode(t *testing.T, tc *testContext, name, nodeName string) {
 		t.Fatalf("%v", err)
 	}
 	t.Cleanup(func() {
-		_ = tc.k8sClient.CoreV1().Pods(ateletNamespace).Delete(context.Background(), name, metav1.DeleteOptions{
+		_ = tc.k8sClient.CoreV1().Pods(installdefaults.SystemNamespace).Delete(context.Background(), name, metav1.DeleteOptions{
 			GracePeriodSeconds: ptr.To[int64](0),
 		})
 	})

cmd/ateapi/internal/controlapi/service.go

@@ -17,6 +17,7 @@ package controlapi
 import (
 	"context"
 	"sync"
+	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
@@ -51,7 +52,8 @@ type VolumePluginRegistry interface {
 	GetPlugin(ctx context.Context, name string) (volume.VolumePluginControlPlane, error)
 }
 
-// NewService creates a service. instruments may be nil; the record helpers no-op.
+// NewService creates a service. actorWorkflowDeadline bounds how long a single
+// Resume/Suspend workflow can run end-to-end. instruments may be nil.
 func NewService(
 	persistence store.Interface,
 	workerCache *workercache.Cache,
@@ -63,6 +65,7 @@ func NewService(
 	dialer *AteletDialer,
 	instruments *Instruments,
 	egressGatewayAddress string,
+	actorWorkflowDeadline time.Duration,
 	volumePlugins map[string]volume.VolumePluginControlPlane,
 ) *Service {
 	s := &Service{
@@ -76,7 +79,7 @@ func NewService(
 		instruments:           instruments,
 		volumePlugins:         volumePlugins,
 	}
-	s.actorWorkflow = NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s)
+	s.actorWorkflow = NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s, actorWorkflowDeadline)
 	return s
 }
 

cmd/ateapi/internal/controlapi/workflow.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/scheduling"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
@@ -79,9 +80,12 @@ type ActorWorkflow struct {
 	instruments          *Instruments
 	egressGatewayAddress string
 	pluginRegistry       VolumePluginRegistry
+	// workflowDeadline is the maximum duration of a single actor workflow.
+	workflowDeadline time.Duration
 }
 
-// NewActorWorkflow creates a new ActorWorkflow. instruments may be nil.
+// NewActorWorkflow creates a new ActorWorkflow. workflowDeadline bounds how
+// long a single Resume/Suspend can run end-to-end; instruments may be nil.
 func NewActorWorkflow(
 	store actorWorkflowStore,
 	workerCache *workercache.Cache,
@@ -93,6 +97,7 @@ func NewActorWorkflow(
 	instruments *Instruments,
 	egressGatewayAddress string,
 	pluginRegistry VolumePluginRegistry,
+	workflowDeadline time.Duration,
 ) *ActorWorkflow {
 	return &ActorWorkflow{
 		store:                store,
@@ -106,6 +111,7 @@ func NewActorWorkflow(
 		instruments:          instruments,
 		egressGatewayAddress: egressGatewayAddress,
 		pluginRegistry:       pluginRegistry,
+		workflowDeadline:     workflowDeadline,
 	}
 }
 
@@ -124,14 +130,17 @@ type actorWorkflowStore interface {
 
 func (w *ActorWorkflow) acquireActorLock(ctx context.Context, actorRef resources.ActorRef) (context.Context, *store.Lock, error) {
 	lockKey := "lock:actor:" + actorRef.Atespace + ":" + actorRef.Name
+	workflowCtx, cancel := context.WithTimeout(ctx, w.workflowDeadline)
 
-	lock, err := w.store.AcquireLock(ctx, lockKey)
+	lock, err := w.store.AcquireLock(workflowCtx, lockKey)
 	if err != nil {
+		cancel()
 		if errors.Is(err, store.ErrLockConflict) {
 			return nil, nil, status.Error(grpcCodes.Aborted, "another operation is in progress for this actor")
 		}
 		return nil, nil, fmt.Errorf("while acquiring lock: %w", err)
 	}
 
+	context.AfterFunc(lock.Context(), cancel)
 	return lock.Context(), lock, nil
 }

cmd/ateapi/internal/controlapi/workflow_lock_test.go

@@ -0,0 +1,49 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlapi
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/ateredis"
+	"github.com/agent-substrate/substrate/internal/resources"
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+func TestAcquireActorLockWorkflowDeadline(t *testing.T) {
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{mr.Addr()}})
+	t.Cleanup(func() { _ = rdb.Close() })
+	w := &ActorWorkflow{store: ateredis.NewPersistence(rdb), workflowDeadline: 20 * time.Millisecond}
+
+	ctx, lock, err := w.acquireActorLock(context.Background(), resources.ActorRef{Atespace: "space", Name: "actor"})
+	if err != nil {
+		t.Fatalf("acquireActorLock: %v", err)
+	}
+	t.Cleanup(lock.Close)
+
+	select {
+	case <-ctx.Done():
+		if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			t.Fatalf("context error = %v, want DeadlineExceeded", ctx.Err())
+		}
+	case <-time.After(time.Second):
+		t.Fatal("workflow context did not reach its deadline")
+	}
+}

cmd/ateapi/internal/controlapi/workflow_suspend_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/ateredis"
@@ -695,7 +696,7 @@ func TestSuspendActor_PausedWithoutLocalSnapshotCrashes(t *testing.T) {
 	}); err != nil {
 		t.Fatalf("add template to indexer: %v", err)
 	}
-	w := NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil)
+	w := NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil, time.Minute)
 
 	seedWorkflowActor(t, ctx, st, resources.ActorRef{Atespace: "team-a", Name: "id1"}, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_PAUSED)
 

cmd/ateapi/internal/controlapi/workflow_testutil_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"slices"
 	"testing"
+	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/internal/resources"
@@ -42,7 +43,7 @@ func newTestActorWorkflow(t *testing.T, st store.Interface, tmplNamespace, tmplN
 	}); err != nil {
 		t.Fatalf("add template to indexer: %v", err)
 	}
-	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil)
+	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil, time.Minute)
 }
 
 // seedWorkflowActor stores an actor with the given state, bound to the given

cmd/ateapi/main.go

@@ -88,6 +88,8 @@ var (
 	drainDelay   = pflag.Duration("drain-delay", 13*time.Second, "How long to keep accepting new work after SIGTERM, before starting the gRPC drain.")
 	drainTimeout = pflag.Duration("drain-timeout", 15*time.Second, "Deadline for the graceful gRPC drain on shutdown. In-flight RPCs still running past it are forcefully cancelled.")
 
+	actorWorkflowDeadline = pflag.Duration("actor-workflow-deadline", 5*time.Minute, "Maximum wall-clock duration of a single Resume/Suspend workflow; raise it for slow image registries.")
+
 	showVersion  = pflag.Bool("version", false, "Print version and exit.")
 	logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.")
 )
@@ -205,7 +207,7 @@ func main() {
 		dialerOpts = append(dialerOpts, controlapi.WithInsecureCredentials())
 	}
 	ateletDialer := controlapi.NewAteletDialer(workerPodInformer.GetIndexer(), ateletPodInformer.GetIndexer(), *ateletClientCredBundle, *podIdentityCACerts, dialerOpts...)
-	sm := controlapi.NewService(persistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, storageClassLister, ateletDialer, instruments, *egressGatewayAddress, volPlugins)
+	sm := controlapi.NewService(persistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, storageClassLister, ateletDialer, instruments, *egressGatewayAddress, *actorWorkflowDeadline, volPlugins)
 
 	actorIdentitySrv := actoridentity.New(actorIdentityJWTIssuer, *actorIDJWTPoolFile, *actorIDCAPoolFile, persistence, workerCache)
 	debugSrv := debugapi.NewService(persistence)
@@ -327,6 +329,7 @@ func logFlagValues(ctx context.Context) {
 		slog.Bool("atelet-insecure", *ateletInsecure),
 		slog.Duration("drain-delay", *drainDelay),
 		slog.Duration("drain-timeout", *drainTimeout),
+		slog.Duration("actor-workflow-deadline", *actorWorkflowDeadline),
 	)
 }