Decouple actor lock TTL from workflow deadline via heartbeat

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

cmd/ateapi/internal/controlapi/functional_test.go

@@ -310,7 +310,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.
@@ -123,9 +136,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,
@@ -134,6 +152,7 @@ func NewActorWorkflow(
 	workerPoolLister listersv1alpha1.WorkerPoolLister,
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	kubeClient kubernetes.Interface,
+	workflowDeadline time.Duration,
 ) *ActorWorkflow {
 	return &ActorWorkflow{
 		store:               store,
@@ -144,6 +163,7 @@ func NewActorWorkflow(
 		sandboxConfigLister: sandboxConfigLister,
 		kubeClient:          kubeClient,
 		secretCache:         newEnvSecretCache(envSecretCacheTTL),
+		workflowDeadline:    workflowDeadline,
 	}
 }
 
@@ -156,9 +176,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
 	}
@@ -186,9 +204,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
 	}
@@ -216,9 +232,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
 	}
@@ -238,27 +252,72 @@ 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, atespace, name string, lockTTL, heartbeatInterval time.Duration) (context.Context, func(), error) {
+	actorID := atespace + ":" + name
+	lockKey := "lock:actor:" + actorID
 	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, actorID, 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

@@ -0,0 +1,124 @@
+// 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/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+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(), "ns", "actor-1", lockTTL, heartbeat)
+	if err != nil {
+		t.Fatalf("acquireActorLock: %v", err)
+	}
+	defer release()
+
+	// Wait through multiple TTLs. If the heartbeat is working the lock key
+	// must still be in Redis — its TTL is being PEXPIRE'd back to `lockTTL`
+	// every `heartbeat`.
+	time.Sleep(4 * lockTTL)
+
+	if !mr.Exists("lock:actor:ns: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)
+
+	lockTTL := 200 * time.Millisecond
+	heartbeat := 30 * time.Millisecond
+
+	ctx, release, err := w.acquireActorLock(context.Background(), "ns", "actor-2", lockTTL, heartbeat)
+	if err != nil {
+		t.Fatalf("acquireActorLock: %v", err)
+	}
+	defer release()
+
+	// Simulate a peer stealing the lock (or the TTL lapsing and someone else
+	// re-acquiring): wipe our key so the next heartbeat refresh's CAS fails.
+	mr.Del("lock:actor:ns: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(), "ns", "actor-3", 200*time.Millisecond, 60*time.Millisecond)
+	if err != nil {
+		t.Fatalf("acquireActorLock: %v", err)
+	}
+
+	if !mr.Exists("lock:actor:ns:actor-3") {
+		t.Fatalf("lock key not in Redis after acquire")
+	}
+	release()
+	if mr.Exists("lock:actor:ns: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(), "ns", "actor-4", 5*time.Second, 1*time.Second)
+	if err != nil {
+		t.Fatalf("first acquireActorLock: %v", err)
+	}
+	defer release()
+
+	_, _, err = w.acquireActorLock(context.Background(), "ns", "actor-4", 5*time.Second, 1*time.Second)
+	if err == nil {
+		t.Fatalf("expected second acquireActorLock to fail")
+	}
+}

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

@@ -838,6 +838,26 @@ func (s *Persistence) ReleaseLock(ctx context.Context, key string, value string)
 	return nil
 }
 
+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
+}
+
 func newCreateMetadata(atespace, name string) *ateapipb.ResourceMetadata {
 	now := timestamppb.Now()
 	return &ateapipb.ResourceMetadata{