ateapi/controlapi: retry worker updates on version conflict in syncer

cmd/ateapi/internal/controlapi/syncer.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"log/slog"
 	"maps"
+	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/internal/resources"
@@ -27,6 +28,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
 )
 
@@ -156,28 +158,68 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po
 		return
 	}
 
-	changed := false
-	if w.Ip != pod.Status.PodIP {
-		// TODO: I don't think this is possible, but handling this case so we can log it just in case we can reproduce it.
-		slog.InfoContext(ctx, "Syncer: updating worker in store (IP changed)", slog.String("worker", pod.Namespace+"/"+pod.Name))
-		w.Ip = pod.Status.PodIP
-		changed = true
-	}
-	if w.SandboxClass != string(pool.Spec.SandboxClass) {
-		slog.InfoContext(ctx, "Syncer: updating worker in store (SandboxClass changed)", slog.String("worker", pod.Namespace+"/"+pod.Name))
-		w.SandboxClass = string(pool.Spec.SandboxClass)
-		changed = true
-	}
-	if !maps.Equal(w.Labels, pool.GetLabels()) {
-		slog.InfoContext(ctx, "Syncer: updating worker in store (labels changed)", slog.String("worker", pod.Namespace+"/"+pod.Name))
-		w.Labels = pool.GetLabels()
-		changed = true
+	// TODO: We perform an inline retry loop here using wait.ExponentialBackoff
+	// as a pragmatic fix for optimistic locking conflicts (store.ErrVersionConflict).
+	// Because WorkerPoolSyncer directly executes in informer event handlers (AddFunc/UpdateFunc),
+	// blocking retries can temporarily stall the informer dispatch thread.
+	// The long-term architectural pattern is to refactor WorkerPoolSyncer to enqueue pod keys
+	// into an asynchronous rate-limiting workqueue (workqueue.RateLimitingInterface) processed by
+	// worker goroutines, allowing clean asynchronous requeuing on version conflicts without blocking.
+	backoff := wait.Backoff{
+		Steps:    5,
+		Duration: 10 * time.Millisecond,
+		Factor:   2.0,
+		Jitter:   1.0,
 	}
+	err = wait.ExponentialBackoff(backoff, func() (bool, error) {
+		if err := ctx.Err(); err != nil {
+			return false, err
+		}
+		if w == nil {
+			var getErr error
+			w, getErr = s.persistence.GetWorker(ctx, pod.Namespace, poolName, pod.Name)
+			if getErr != nil {
+				if errors.Is(getErr, store.ErrNotFound) {
+					return true, nil // worker disappeared, stop retrying
+				}
+				return false, getErr
+			}
+		}
 
-	if changed {
-		if err = s.persistence.UpdateWorker(ctx, w, w.Version); err != nil {
-			slog.ErrorContext(ctx, "Failed to update worker in store", slog.Any("err", err))
+		changed := false
+		if w.Ip != pod.Status.PodIP {
+			// TODO: I don't think this is possible, but handling this case so we can log it just in case we can reproduce it.
+			slog.InfoContext(ctx, "Syncer: updating worker in store (IP changed)", slog.String("worker", pod.Namespace+"/"+pod.Name))
+			w.Ip = pod.Status.PodIP
+			changed = true
+		}
+		if w.SandboxClass != string(pool.Spec.SandboxClass) {
+			slog.InfoContext(ctx, "Syncer: updating worker in store (SandboxClass changed)", slog.String("worker", pod.Namespace+"/"+pod.Name))
+			w.SandboxClass = string(pool.Spec.SandboxClass)
+			changed = true
+		}
+		if !maps.Equal(w.Labels, pool.GetLabels()) {
+			slog.InfoContext(ctx, "Syncer: updating worker in store (labels changed)", slog.String("worker", pod.Namespace+"/"+pod.Name))
+			w.Labels = pool.GetLabels()
+			changed = true
 		}
+
+		if !changed {
+			return true, nil
+		}
+
+		if err := s.persistence.UpdateWorker(ctx, w, w.Version); err != nil {
+			if errors.Is(err, store.ErrVersionConflict) {
+				slog.InfoContext(ctx, "Syncer: version conflict updating worker, retrying", slog.String("worker", pod.Namespace+"/"+pod.Name))
+				w = nil // force re-fetch on next retry attempt
+				return false, nil
+			}
+			return false, err
+		}
+		return true, nil
+	})
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to update worker in store after retries", slog.Any("err", err))
 	}
 }
 

cmd/ateapi/internal/controlapi/syncer_test.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -38,9 +39,17 @@ import (
 
 // setupSyncerTest sets up a real store with fake Redis and a fake K8s client with informer.
 func setupSyncerTest(t *testing.T, ctx context.Context, initPools ...*atev1alpha1.WorkerPool) (store.Interface, *fake.Clientset, *atefake.Clientset, func()) {
+	return setupSyncerTestWithStore(t, ctx, nil, initPools...)
+}
+
+func setupSyncerTestWithStore(t *testing.T, ctx context.Context, wrapStore func(store.Interface) store.Interface, initPools ...*atev1alpha1.WorkerPool) (store.Interface, *fake.Clientset, *atefake.Clientset, func()) {
 	t.Helper()
 
 	persistence, cleanup := storetest.SetupTestStore(t)
+	if wrapStore != nil {
+		persistence = wrapStore(persistence)
+	}
+
 
 	fakeK8s := fake.NewSimpleClientset()
 	workerFactory, workerInformer := WorkerPodInformer(fakeK8s)
@@ -361,3 +370,143 @@ func TestSyncer_OmittedFields(t *testing.T) {
 		t.Fatalf("Worker state check failed: %v", err)
 	}
 }
+
+type conflictStore struct {
+	store.Interface
+	conflictTriggered atomic.Bool
+	onUpdate          func(ctx context.Context, worker *ateapipb.Worker)
+}
+
+func (c *conflictStore) UpdateWorker(ctx context.Context, worker *ateapipb.Worker, expectedVersion int64) error {
+	if c.onUpdate != nil && c.conflictTriggered.CompareAndSwap(false, true) {
+		c.onUpdate(ctx, worker)
+	}
+	return c.Interface.UpdateWorker(ctx, worker, expectedVersion)
+}
+
+func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	ns := "ns-syncer-conflict"
+	podName := "worker-unit-conflict"
+	poolName := "pool-conflict"
+
+	pool := &atev1alpha1.WorkerPool{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      poolName,
+			Namespace: ns,
+			Labels:    map[string]string{"foo": "bar"},
+		},
+		Spec: atev1alpha1.WorkerPoolSpec{
+			SandboxClass: "gvisor",
+		},
+	}
+
+	var cs *conflictStore
+	persistence, fakeK8s, fakeAte, cleanup := setupSyncerTestWithStore(t, ctx, func(s store.Interface) store.Interface {
+		cs = &conflictStore{Interface: s}
+		return cs
+	}, pool)
+	defer cleanup()
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      podName,
+			Namespace: ns,
+			UID:       "11111111-2222-3333-4444-555555555555",
+			Labels: map[string]string{
+				workerPodLabel: poolName,
+			},
+		},
+		Spec: corev1.PodSpec{
+			NodeName:   "node1",
+			Containers: []corev1.Container{{Name: "main", Image: "nginx"}},
+		},
+		Status: corev1.PodStatus{
+			Phase:  corev1.PodRunning,
+			PodIP:  "10.0.0.1",
+			PodIPs: []corev1.PodIP{{IP: "10.0.0.1"}},
+		},
+	}
+
+	_, err := fakeK8s.CoreV1().Pods(ns).Create(context.Background(), pod, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("failed to create pod: %v", err)
+	}
+
+	err = wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) {
+		w, err := persistence.GetWorker(ctx, ns, poolName, podName)
+		if err != nil {
+			if errors.Is(err, store.ErrNotFound) {
+				return false, nil
+			}
+			return false, err
+		}
+		return w.Ip == "10.0.0.1", nil
+	})
+	if err != nil {
+		t.Fatalf("Worker state check failed: %v", err)
+	}
+
+	// Update the pool SandboxClass in K8s (a mutable worker field).
+	updatedPool, err := fakeAte.ApiV1alpha1().WorkerPools(ns).Get(context.Background(), poolName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("failed to get pool: %v", err)
+	}
+	updatedPool.Spec.SandboxClass = "microvm"
+	if _, err := fakeAte.ApiV1alpha1().WorkerPools(ns).Update(context.Background(), updatedPool, metav1.UpdateOptions{}); err != nil {
+		t.Fatalf("failed to update pool: %v", err)
+	}
+
+	// Wait until the WorkerPool informer cache reflects the updated SandboxClass before triggering the pod update.
+	err = wait.PollUntilContextTimeout(context.Background(), 50*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) {
+		p, err := fakeAte.ApiV1alpha1().WorkerPools(ns).Get(ctx, poolName, metav1.GetOptions{})
+		if err != nil {
+			return false, nil
+		}
+		return p.Spec.SandboxClass == "microvm", nil
+	})
+	if err != nil {
+		t.Fatalf("pool informer cache failed to update: %v", err)
+	}
+
+	// Configure conflictStore to inject a concurrent version bump in Redis when syncWorkerToStore calls UpdateWorker.
+	cs.onUpdate = func(c context.Context, w *ateapipb.Worker) {
+		if cw, err := cs.Interface.GetWorker(c, ns, poolName, podName); err == nil {
+			cw.NodeName = "node2"
+			_ = cs.Interface.UpdateWorker(c, cw, cw.Version)
+		}
+	}
+
+	// Touch the pod ONCE in K8s so WorkerPoolSyncer triggers syncWorkerToStore. It will attempt an update
+	// with its cached/old worker version and encounter ErrVersionConflict (injected by conflictStore),
+	// requiring it to retry by fetching the latest version from Redis.
+	updatedPod, err := fakeK8s.CoreV1().Pods(ns).Get(context.Background(), podName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("failed to get pod: %v", err)
+	}
+	if updatedPod.Annotations == nil {
+		updatedPod.Annotations = make(map[string]string)
+	}
+	updatedPod.Annotations["trigger"] = "update"
+	if _, err := fakeK8s.CoreV1().Pods(ns).Update(context.Background(), updatedPod, metav1.UpdateOptions{}); err != nil {
+		t.Fatalf("failed to update pod: %v", err)
+	}
+
+	// Verify that the worker in Redis eventually gets updated to the new SandboxClass despite the version conflict.
+	err = wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) {
+		w, err := persistence.GetWorker(ctx, ns, poolName, podName)
+		if err != nil {
+			if errors.Is(err, store.ErrNotFound) {
+				return false, nil
+			}
+			return false, err
+		}
+		return w.SandboxClass == "microvm" && w.NodeName == "node2", nil
+	})
+	if err != nil {
+		t.Fatalf("Worker failed to update SandboxClass after version conflict: %v", err)
+	}
+}
+