fix(ateapi): drop worker watch events instead of stalling pub/sub

cmd/ateapi/internal/controlapi/functional_test.go

@@ -295,7 +295,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 	rdb := redis.NewClusterClient(&redis.ClusterOptions{
 		Addrs: []string{mr.Addr()},
 	})
-	persistence := ateredis.NewPersistence(rdb)
+	persistence := ateredis.NewPersistence(rdb, nil)
 
 	// 2. Initialize Clientsets using global cfg
 	k8sClient, err := kubernetes.NewForConfig(cfg)

cmd/ateapi/internal/controlapi/workflow_suspend_test.go

@@ -226,7 +226,7 @@ func newTestPersistence(t *testing.T) store.Interface {
 	t.Cleanup(mr.Close)
 	rdb := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{mr.Addr()}})
 	t.Cleanup(func() { rdb.Close() }) //nolint:errcheck // test cleanup
-	return ateredis.NewPersistence(rdb)
+	return ateredis.NewPersistence(rdb, nil)
 }
 
 // newDanglingDialer returns a dialer whose informer cache has no pods, so

cmd/ateapi/internal/store/atepg/atepg.go

@@ -1506,7 +1506,8 @@ func (p *Persistence) WatchWorkers(ctx context.Context) (*store.WorkerWatch, err
 			}
 		}
 	}()
-	return store.NewWorkerWatch(ch, cancel), nil
+	// nil invalidation: this watch blocks on a full buffer rather than dropping.
+	return store.NewWorkerWatch(ch, nil, cancel), nil
 }
 
 // --- Workflow locks ---

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

@@ -57,6 +57,7 @@ import (
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"github.com/google/uuid"
 	"github.com/redis/go-redis/v9"
+	"go.opentelemetry.io/otel/metric"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -79,18 +80,34 @@ type redisClient interface {
 
 // Persistence is a service that stores information about applications in Redis.
 type Persistence struct {
-	rdb     redisClient
-	lockTTL time.Duration
+	rdb           redisClient
+	lockTTL       time.Duration
+	droppedEvents metric.Int64Counter
 }
 
 var _ store.Interface = (*Persistence)(nil)
 
-// NewPersistence creates a new Persistence.
-func NewPersistence(redisClient *redis.ClusterClient) *Persistence {
-	return &Persistence{
+const droppedEventsMetric = "ate.store.worker_watch.dropped_events"
+
+// NewPersistence creates a new Persistence. meter may be nil to disable metrics.
+func NewPersistence(redisClient *redis.ClusterClient, meter metric.Meter) *Persistence {
+	p := &Persistence{
 		rdb:     redisClient,
 		lockTTL: defaultLockTTL,
 	}
+	if meter != nil {
+		counter, err := meter.Int64Counter(
+			droppedEventsMetric,
+			metric.WithUnit("{event}"),
+			metric.WithDescription("Worker watch events dropped because the consumer buffer was full."),
+		)
+		if err != nil {
+			slog.Error("Failed to register worker watch dropped-events counter", "metric", droppedEventsMetric, "error", err)
+		} else {
+			p.droppedEvents = counter
+		}
+	}
+	return p
 }
 
 // actorDBKey returns the Redis key an actor is stored under. The encoding is
@@ -659,6 +676,7 @@ func (s *Persistence) WatchWorkers(ctx context.Context) (*store.WorkerWatch, err
 		return nil, fmt.Errorf("while confirming worker subscription: %w", err)
 	}
 	ch := make(chan store.WorkerEvent, 128)
+	invalidated := make(chan struct{}, 1)
 	go func() {
 		defer close(ch)
 		defer pubsub.Close()
@@ -678,13 +696,22 @@ func (s *Persistence) WatchWorkers(ctx context.Context) (*store.WorkerWatch, err
 				}
 				select {
 				case ch <- event:
-				case <-watchCtx.Done():
-					return
+				default:
+					// Blocking here would stall the pub/sub socket and drop the
+					// connection; drop the event and coalesce an invalidation.
+					if s.droppedEvents != nil {
+						s.droppedEvents.Add(ctx, 1)
+					}
+					select {
+					case invalidated <- struct{}{}:
+						slog.WarnContext(ctx, "worker watch buffer full; dropping events, invalidation signaled")
+					default:
+					}
 				}
 			}
 		}
 	}()
-	return store.NewWorkerWatch(ch, cancel), nil
+	return store.NewWorkerWatch(ch, invalidated, cancel), nil
 }
 
 // DebugClearAll flushes all data from Redis.

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

@@ -28,6 +28,9 @@ import (
 	"github.com/alicebob/miniredis/v2"
 	"github.com/google/go-cmp/cmp"
 	"github.com/redis/go-redis/v9"
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/testing/protocmp"
@@ -38,6 +41,10 @@ import (
 )
 
 func setupTest(t *testing.T) (*miniredis.Miniredis, *Persistence, context.Context) {
+	return setupTestWithMeter(t, nil)
+}
+
+func setupTestWithMeter(t *testing.T, meter metric.Meter) (*miniredis.Miniredis, *Persistence, context.Context) {
 	mr, err := miniredis.Run()
 	if err != nil {
 		t.Fatalf("failed to start miniredis: %v", err)
@@ -50,7 +57,7 @@ func setupTest(t *testing.T) (*miniredis.Miniredis, *Persistence, context.Contex
 		Addrs: []string{mr.Addr()},
 	})
 	t.Cleanup(func() { rdb.Close() })
-	return mr, NewPersistence(rdb), t.Context()
+	return mr, NewPersistence(rdb, meter), t.Context()
 }
 
 // testAtespace is the atespace used by tests that create a single actor. Actors
@@ -3236,3 +3243,89 @@ func TestDeleteActorTemplate_VersionInOtherAtespace_NotBlocking(t *testing.T) {
 		t.Errorf("DeleteActorTemplate = %v, want nil: the only version lives in team-b", err)
 	}
 }
+
+func droppedEventsCount(t *testing.T, reader *sdkmetric.ManualReader) int64 {
+	t.Helper()
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(context.Background(), &rm); err != nil {
+		t.Fatalf("collecting metrics: %v", err)
+	}
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name != droppedEventsMetric {
+				continue
+			}
+			var total int64
+			for _, dp := range m.Data.(metricdata.Sum[int64]).DataPoints {
+				total += dp.Value
+			}
+			return total
+		}
+	}
+	return 0
+}
+
+func TestWatchWorkersOverflowSignalsInvalidation(t *testing.T) {
+	reader := sdkmetric.NewManualReader()
+	meter := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test")
+	_, s, ctx := setupTestWithMeter(t, meter)
+
+	watch, err := s.WatchWorkers(ctx)
+	if err != nil {
+		t.Fatalf("WatchWorkers failed: %v", err)
+	}
+	defer watch.Close()
+
+	for i := 0; i < 200; i++ {
+		if err := s.CreateWorker(ctx, &ateapipb.Worker{
+			WorkerNamespace: "default",
+			WorkerPool:      "pool-1",
+			WorkerPod:       fmt.Sprintf("pod-%d", i),
+		}); err != nil {
+			t.Fatalf("CreateWorker %d failed: %v", i, err)
+		}
+	}
+
+	select {
+	case <-watch.Invalidated:
+	case <-time.After(10 * time.Second):
+		t.Fatal("no invalidation signal after overflowing the watch buffer")
+	}
+
+	if got := droppedEventsCount(t, reader); got <= 0 {
+		t.Errorf("%s = %d, want > 0 after overflow", droppedEventsMetric, got)
+	}
+
+	for draining := true; draining; {
+		select {
+		case _, ok := <-watch.Events:
+			if !ok {
+				t.Fatal("watch events channel closed after overflow; the subscription must survive")
+			}
+		case <-time.After(200 * time.Millisecond):
+			draining = false
+		}
+	}
+
+	if err := s.CreateWorker(ctx, &ateapipb.Worker{
+		WorkerNamespace: "default",
+		WorkerPool:      "pool-1",
+		WorkerPod:       "pod-after-overflow",
+	}); err != nil {
+		t.Fatalf("CreateWorker after overflow failed: %v", err)
+	}
+	deadline := time.After(10 * time.Second)
+	for {
+		select {
+		case event, ok := <-watch.Events:
+			if !ok {
+				t.Fatal("watch events channel closed after overflow")
+			}
+			if event.Worker.GetWorkerPod() == "pod-after-overflow" {
+				return
+			}
+		case <-deadline:
+			t.Fatal("event published after overflow never delivered")
+		}
+	}
+}

cmd/ateapi/internal/store/store.go

@@ -280,15 +280,18 @@ type WorkerEvent struct {
 type WorkerWatch struct {
 	// Events delivers worker state changes until the watch is torn down.
 	Events <-chan WorkerEvent
+	// Invalidated signals that events were dropped and the consumer should
+	// relist. Signals coalesce; nil for stores that never drop events.
+	Invalidated <-chan struct{}
 	// stop releases the subscription backing Events. It is a context.CancelFunc,
 	// so it is safe to call multiple times.
 	stop context.CancelFunc
 }
 
-// NewWorkerWatch builds a WorkerWatch from an events channel and the cancel
-// func that tears down its subscription.
-func NewWorkerWatch(events <-chan WorkerEvent, stop context.CancelFunc) *WorkerWatch {
-	return &WorkerWatch{Events: events, stop: stop}
+// NewWorkerWatch builds a WorkerWatch from an events channel, an optional
+// invalidation channel, and the cancel func that tears down its subscription.
+func NewWorkerWatch(events <-chan WorkerEvent, invalidated <-chan struct{}, stop context.CancelFunc) *WorkerWatch {
+	return &WorkerWatch{Events: events, Invalidated: invalidated, stop: stop}
 }
 
 // Close releases the subscription. Safe to call multiple times.

cmd/ateapi/internal/store/storetest/storetest.go

@@ -37,7 +37,7 @@ func SetupTestStore(t *testing.T) (store.Interface, func()) {
 		Addrs: []string{mr.Addr()},
 	})
 
-	persistence := ateredis.NewPersistence(rdb)
+	persistence := ateredis.NewPersistence(rdb, nil)
 
 	cleanup := func() {
 		rdb.Close()

cmd/ateapi/internal/workercache/workercache.go

@@ -34,6 +34,11 @@ import (
 // relistPageSize is the page size used for the relist.
 const relistPageSize = 1000
 
+const (
+	minInvalidationRetry = 100 * time.Millisecond
+	maxInvalidationRetry = 5 * time.Second
+)
+
 // Cache maintains an in-memory snapshot of all workers.
 //
 // TODO: add metrics — at minimum a gauge for worker count, a counter for
@@ -143,6 +148,30 @@ func (c *Cache) relist(ctx context.Context) error {
 func (c *Cache) watchEvents(ctx context.Context, watch *store.WorkerWatch) {
 	ticker := time.NewTicker(c.relistInterval)
 	defer ticker.Stop()
+
+	// Signal consumed and backlog discarded: a failed relist has nothing to retrigger it, so retry ourselves.
+	var retry <-chan time.Time
+	retryDelay := minInvalidationRetry
+	correct := func() {
+		// Discard events buffered before the drop so a stale event can't roll
+		// back the relisted snapshot.
+		for draining := true; draining; {
+			select {
+			case _, ok := <-watch.Events:
+				draining = ok
+			default:
+				draining = false
+			}
+		}
+		if err := c.relist(ctx); err != nil {
+			slog.WarnContext(ctx, "worker cache: invalidation relist failed, retrying", slog.Any("err", err), slog.Duration("in", retryDelay))
+			retry = time.After(retryDelay)
+			retryDelay = min(retryDelay*2, maxInvalidationRetry)
+			return
+		}
+		retry, retryDelay = nil, minInvalidationRetry
+	}
+
 	for {
 		select {
 		case event, ok := <-watch.Events:
@@ -157,13 +186,22 @@ func (c *Cache) watchEvents(ctx context.Context, watch *store.WorkerWatch) {
 				if watch == nil {
 					return // context cancelled
 				}
+				// resync already relisted; a pending correction would only drain the new watch.
+				retry, retryDelay = nil, minInvalidationRetry
 				c.ready.Store(true)
 			} else {
 				c.applyEvent(event)
 			}
+		case <-watch.Invalidated:
+			slog.WarnContext(ctx, "worker cache: watch events dropped, relisting")
+			correct()
+		case <-retry:
+			correct()
 		case <-ticker.C:
 			if err := c.relist(ctx); err != nil {
 				slog.WarnContext(ctx, "worker cache: periodic relist failed", slog.Any("err", err))
+			} else {
+				retry, retryDelay = nil, minInvalidationRetry
 			}
 		case <-ctx.Done():
 			c.ready.Store(false)