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.

Author: jjamroga

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

@@ -64,8 +64,26 @@ spec:
                   description: A single application container that you want to run
                     within a WorkerPool.
                   properties:
+                    args:
+                      description: |-
+                        Arguments to the entrypoint. Not executed within a shell. The container
+                        image's CMD is used if this is not provided (unless command is set,
+                        which discards the image's CMD).
+
+                        Unlike Kubernetes, variable references $(VAR_NAME) are NOT expanded.
+                      items:
+                        type: string
+                      maxItems: 64
+                      type: array
+                      x-kubernetes-list-type: atomic
                     command:
-                      description: Entrypoint array. Not executed within a shell.
+                      description: |-
+                        Entrypoint array. Not executed within a shell. The container image's
+                        ENTRYPOINT is used if this is not provided; if it is provided, the
+                        image's ENTRYPOINT and CMD are both ignored and the process argv is
+                        command + args.
+
+                        Unlike Kubernetes, variable references $(VAR_NAME) are NOT expanded.
                       items:
                         type: string
                       maxItems: 64

cmd/ateapi/internal/controlapi/functional_test.go

@@ -314,7 +314,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 	}
 
 	dialer := NewAteletDialer(workerInformer.GetIndexer(), ateletInformer.GetIndexer())
-	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, k8sClient)
+	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, k8sClient, 30*time.Second)
 
 	// 5. Start REAL gRPC Server for ATE API
 	grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ateinterceptors.ServerUnaryInterceptor))

cmd/ateapi/internal/controlapi/service.go

@@ -15,6 +15,8 @@
 package controlapi
 
 import (
+	"time"
+
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
 	listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1"
@@ -34,7 +36,8 @@ type Service struct {
 
 var _ ateapipb.ControlServer = (*Service)(nil)
 
-// NewService creates a service.
+// NewService creates a service. actorWorkflowDeadline bounds how long a single
+// Resume/Suspend workflow can run end-to-end.
 func NewService(
 	persistence store.Interface,
 	workerCache *workercache.Cache,
@@ -43,13 +46,14 @@ func NewService(
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	dialer *AteletDialer,
 	kubeClient kubernetes.Interface,
+	actorWorkflowDeadline time.Duration,
 ) *Service {
 	s := &Service{
 		persistence:         persistence,
 		actorTemplateLister: actorTemplateLister,
 		workerPoolLister:    workerPoolLister,
 		dialer:              dialer,
-		actorWorkflow:       NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient),
+		actorWorkflow:       NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient, actorWorkflowDeadline),
 	}
 
 	return s

cmd/ateapi/internal/controlapi/workflow.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
@@ -33,6 +34,18 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
+// actorLockTTL is the Redis TTL on the per-actor workflow lock. It bounds how
+// long a peer must wait to retry an actor after this process crashes mid-workflow.
+const actorLockTTL = 30 * time.Second
+
+// actorLockHeartbeatInterval is how often the heartbeat refreshes the lock.
+// Chosen so we get ~3 attempts before the TTL would otherwise lapse.
+const actorLockHeartbeatInterval = actorLockTTL / 3
+
+// errLostActorLock is the context cause set when the heartbeat can no longer
+// keep the actor lock alive (peer stole it, or Redis returned an error).
+var errLostActorLock = errors.New("lost actor lock during workflow")
+
 // WorkflowStep represents a single, idempotent operation in a workflow graph.
 // Params is the immutable parameters used to start the workflow.
 // Context is the mutable context fetched or modified during execution.
@@ -138,9 +151,14 @@ type ActorWorkflow struct {
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister
 	kubeClient          kubernetes.Interface
 	secretCache         *envSecretCache
+	// workflowDeadline is the maximum duration of a single Resume/Suspend
+	// workflow. The lock is kept alive across this duration by a heartbeat,
+	// independent of actorLockTTL.
+	workflowDeadline time.Duration
 }
 
-// NewActorWorkflow creates a new ActorWorkflow.
+// NewActorWorkflow creates a new ActorWorkflow. workflowDeadline bounds how
+// long a single Resume/Suspend can run end-to-end.
 func NewActorWorkflow(
 	store store.Interface,
 	workerCache *workercache.Cache,
@@ -149,6 +167,7 @@ func NewActorWorkflow(
 	workerPoolLister listersv1alpha1.WorkerPoolLister,
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	kubeClient kubernetes.Interface,
+	workflowDeadline time.Duration,
 ) *ActorWorkflow {
 	return &ActorWorkflow{
 		store:               store,
@@ -159,6 +178,7 @@ func NewActorWorkflow(
 		sandboxConfigLister: sandboxConfigLister,
 		kubeClient:          kubeClient,
 		secretCache:         newEnvSecretCache(envSecretCacheTTL),
+		workflowDeadline:    workflowDeadline,
 	}
 }
 
@@ -171,9 +191,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, atespace, name string,
 	}
 	state := &ResumeState{}
 
-	// Acquire lock and get the timeout context for the workflow
-	// Lock TTL is 30 seconds, with 2 seconds padding for workflow timeout
-	ctx, releaseLock, err := w.acquireActorLock(ctx, atespace, name, 30*time.Second, 2*time.Second)
+	ctx, releaseLock, err := w.acquireActorLock(ctx, atespace+":"+name, actorLockTTL, actorLockHeartbeatInterval)
 	if err != nil {
 		return nil, err
 	}
@@ -201,9 +219,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, atespace, name string)
 	}
 	state := &SuspendState{}
 
-	// Acquire lock and get the timeout context for the workflow
-	// Lock TTL is 30 seconds, with 2 seconds padding for workflow timeout
-	ctx, releaseLock, err := w.acquireActorLock(ctx, atespace, name, 30*time.Second, 2*time.Second)
+	ctx, releaseLock, err := w.acquireActorLock(ctx, atespace+":"+name, actorLockTTL, actorLockHeartbeatInterval)
 	if err != nil {
 		return nil, err
 	}
@@ -231,9 +247,7 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, atespace, name string) (
 	}
 	state := &PauseState{}
 
-	// Acquire lock and get the timeout context for the workflow
-	// Lock TTL is 30 seconds, with 2 seconds padding for workflow timeout
-	ctx, releaseLock, err := w.acquireActorLock(ctx, atespace, name, 30*time.Second, 2*time.Second)
+	ctx, releaseLock, err := w.acquireActorLock(ctx, atespace+":"+name, actorLockTTL, actorLockHeartbeatInterval)
 	if err != nil {
 		return nil, err
 	}
@@ -253,27 +267,71 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, atespace, name string) (
 	return state.Actor, nil
 }
 
-func (w *ActorWorkflow) acquireActorLock(ctx context.Context, atespace, name string, ttl time.Duration, padding time.Duration) (context.Context, func(), error) {
-	lockKey := "lock:actor:" + atespace + ":" + name
+// acquireActorLock takes the per-actor workflow lock and returns a workflow
+// context bounded by w.workflowDeadline. A background heartbeat keeps the lock
+// alive — independent of lockTTL — for as long as the workflow runs. If the
+// heartbeat fails (Redis error or another peer stole the lock) the returned
+// context is cancelled with errLostActorLock as the cause, and in-flight steps
+// will see ctx.Err() and unwind. The returned release function stops the
+// heartbeat, waits for it to exit, then best-effort releases the lock.
+func (w *ActorWorkflow) acquireActorLock(ctx context.Context, id string, lockTTL, heartbeatInterval time.Duration) (context.Context, func(), error) {
+	lockKey := "lock:actor:" + id
 	lockValue := uuid.New().String()
 
-	// Create a child context for the workflow that expires BEFORE the lock
-	workflowTimeout := ttl - padding
-	workflowCtx, cancel := context.WithTimeout(ctx, workflowTimeout)
-
-	acquired, err := w.store.AcquireLock(workflowCtx, lockKey, lockValue, ttl)
+	acquired, err := w.store.AcquireLock(ctx, lockKey, lockValue, lockTTL)
 	if err != nil {
-		cancel()
 		return nil, nil, fmt.Errorf("while acquiring lock: %w", err)
 	}
 	if !acquired {
-		cancel()
 		return nil, nil, status.Error(grpcCodes.Aborted, "another operation is in progress for this actor")
 	}
 
-	return workflowCtx, func() {
-		cancel()
+	cancellableCtx, cancelCause := context.WithCancelCause(ctx)
+	workflowCtx, cancelDeadline := context.WithTimeout(cancellableCtx, w.workflowDeadline)
+
+	heartbeatDone := make(chan struct{})
+	go w.runLockHeartbeat(workflowCtx, lockKey, lockValue, id, lockTTL, heartbeatInterval, cancelCause, heartbeatDone)
+
+	release := func() {
+		cancelDeadline()
+		cancelCause(context.Canceled)
+		<-heartbeatDone
 		// Use context.Background() to ensure the lock is released even if the workflow context was canceled.
 		w.store.ReleaseLock(context.Background(), lockKey, lockValue) //nolint:errcheck // best-effort release; the lock TTL is the safety net.
-	}, nil
+	}
+	return workflowCtx, release, nil
+}
+
+// runLockHeartbeat refreshes the actor lock on a ticker until ctx is done. If
+// a refresh fails or returns false (we no longer own the lock), it cancels the
+// workflow context with errLostActorLock so workflow steps tear down promptly.
+func (w *ActorWorkflow) runLockHeartbeat(ctx context.Context, lockKey, lockValue, actorID string, lockTTL, heartbeatInterval time.Duration, cancelCause context.CancelCauseFunc, done chan<- struct{}) {
+	defer close(done)
+	ticker := time.NewTicker(heartbeatInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			ok, err := w.store.RefreshLock(ctx, lockKey, lockValue, lockTTL)
+			if err != nil {
+				// If ctx was cancelled out from under us we're already tearing
+				// down — no need to set a misleading cause.
+				if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+					slog.WarnContext(ctx, "Lock heartbeat failed; cancelling workflow",
+						slog.String("actor_id", actorID),
+						slog.String("err", err.Error()))
+					cancelCause(fmt.Errorf("%w: %w", errLostActorLock, err))
+				}
+				return
+			}
+			if !ok {
+				slog.WarnContext(ctx, "Actor lock no longer owned; cancelling workflow",
+					slog.String("actor_id", actorID))
+				cancelCause(errLostActorLock)
+				return
+			}
+		}
+	}
 }

cmd/ateapi/internal/controlapi/workflow_test.go

@@ -21,6 +21,9 @@ import (
 	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/ateredis"
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -205,3 +208,93 @@ func TestRunWorkflow_RetryOnPersistenceConflict(t *testing.T) {
 		}
 	})
 }
+
+func newLockTestWorkflow(t *testing.T) (*miniredis.Miniredis, *ActorWorkflow) {
+	t.Helper()
+	mr, err := miniredis.Run()
+	if err != nil {
+		t.Fatalf("miniredis.Run: %v", err)
+	}
+	t.Cleanup(mr.Close)
+	rdb := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{mr.Addr()}})
+	return mr, &ActorWorkflow{
+		store:            ateredis.NewPersistence(rdb),
+		workflowDeadline: 30 * time.Second,
+	}
+}
+
+func TestAcquireActorLock_HeartbeatKeepsLockAlivePastTTL(t *testing.T) {
+	mr, w := newLockTestWorkflow(t)
+
+	lockTTL := 150 * time.Millisecond
+	heartbeat := 40 * time.Millisecond
+
+	ctx, release, err := w.acquireActorLock(context.Background(), "actor-1", lockTTL, heartbeat)
+	if err != nil {
+		t.Fatalf("acquireActorLock: %v", err)
+	}
+	defer release()
+
+	time.Sleep(4 * lockTTL)
+
+	if !mr.Exists("lock:actor:actor-1") {
+		t.Fatalf("lock key disappeared from Redis despite heartbeat; ctx err=%v cause=%v", ctx.Err(), context.Cause(ctx))
+	}
+	if ctx.Err() != nil {
+		t.Fatalf("workflow ctx cancelled while heartbeat was healthy: err=%v cause=%v", ctx.Err(), context.Cause(ctx))
+	}
+}
+
+func TestAcquireActorLock_LostLockCancelsWorkflow(t *testing.T) {
+	mr, w := newLockTestWorkflow(t)
+
+	ctx, release, err := w.acquireActorLock(context.Background(), "actor-2", 200*time.Millisecond, 30*time.Millisecond)
+	if err != nil {
+		t.Fatalf("acquireActorLock: %v", err)
+	}
+	defer release()
+
+	mr.Del("lock:actor:actor-2")
+
+	select {
+	case <-ctx.Done():
+	case <-time.After(2 * time.Second):
+		t.Fatalf("workflow ctx was not cancelled after lock was lost")
+	}
+
+	if cause := context.Cause(ctx); !errors.Is(cause, errLostActorLock) {
+		t.Errorf("context.Cause = %v, want errLostActorLock", cause)
+	}
+}
+
+func TestAcquireActorLock_ReleaseRemovesLock(t *testing.T) {
+	mr, w := newLockTestWorkflow(t)
+
+	_, release, err := w.acquireActorLock(context.Background(), "actor-3", 200*time.Millisecond, 60*time.Millisecond)
+	if err != nil {
+		t.Fatalf("acquireActorLock: %v", err)
+	}
+
+	if !mr.Exists("lock:actor:actor-3") {
+		t.Fatalf("lock key not in Redis after acquire")
+	}
+	release()
+	if mr.Exists("lock:actor:actor-3") {
+		t.Errorf("lock key still in Redis after release")
+	}
+}
+
+func TestAcquireActorLock_ConflictReturnsAborted(t *testing.T) {
+	_, w := newLockTestWorkflow(t)
+
+	_, release, err := w.acquireActorLock(context.Background(), "actor-4", 5*time.Second, time.Second)
+	if err != nil {
+		t.Fatalf("first acquireActorLock: %v", err)
+	}
+	defer release()
+
+	_, _, err = w.acquireActorLock(context.Background(), "actor-4", 5*time.Second, time.Second)
+	if err == nil {
+		t.Fatalf("expected second acquireActorLock to fail")
+	}
+}

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"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
@@ -41,7 +42,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)
+	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, time.Minute)
 }
 
 // seedWorkflowActor stores an actor with the given status, bound to the given

cmd/ateapi/internal/store/ateredis/ateredis.go

@@ -809,3 +809,23 @@ func newUpdateMetadata(current *ateapipb.ResourceMetadata) *ateapipb.ResourceMet
 	next.UpdateTime = timestamppb.Now()
 	return next
 }
+
+func (s *Persistence) RefreshLock(ctx context.Context, key string, value string, ttl time.Duration) (bool, error) {
+	var luaRefresh = redis.NewScript(`
+		if redis.call("get", KEYS[1]) == ARGV[1] then
+			return redis.call("pexpire", KEYS[1], ARGV[2])
+		else
+			return 0
+		end
+	`)
+
+	res, err := luaRefresh.Run(ctx, s.rdb, []string{key}, value, ttl.Milliseconds()).Result()
+	if err != nil {
+		return false, fmt.Errorf("while refreshing lock for %q with value %q: %w", key, value, err)
+	}
+	n, ok := res.(int64)
+	if !ok {
+		return false, fmt.Errorf("while refreshing lock for %q: unexpected result type %T", key, res)
+	}
+	return n == 1, nil
+}