1 files changed, 37 insertions, 0 deletions
diff --git a/internal/controlplane/manager_test.go b/internal/controlplane/manager_test.go
index ee78a00..b292060 100644
--- a/internal/controlplane/manager_test.go
+++ b/internal/controlplane/manager_test.go
@@ -192,6 +192,43 @@ func TestSubscriptionReconnectsAfterChannelCloses(t *testing.T) {
 	}
 }
 
+func TestSubscriptionMessageRestoresConnectedStateAfterPublishFailure(t *testing.T) {
+	store := newFakeManagerStore(1)
+	store.redisEnabled = true
+	store.publishErr = errors.New("redis unavailable")
+	messages := make(chan ChangeMessage, 1)
+	store.subscribe = func(_ context.Context, _ int64) (<-chan ChangeMessage, func() error, error) {
+		return messages, func() error { return nil }, nil
+	}
+	manager := newTestManager(store, slog.New(slog.NewTextHandler(&safeLogBuffer{}, nil)), 10*time.Millisecond)
+	if _, err := manager.Reload(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		manager.Run(ctx)
+		close(done)
+	}()
+	waitUntil(t, time.Second, manager.RedisConnected)
+	if err := manager.AfterMutation(context.Background(), 1, "model", "model-1"); err != nil {
+		t.Fatal(err)
+	}
+	waitUntil(t, time.Second, func() bool { return !manager.RedisConnected() })
+
+	store.snapshot.Store(Snapshot{Generation: 2})
+	messages <- ChangeMessage{Payload: `{"generation":2,"resource":"model"}`}
+	waitUntil(t, time.Second, func() bool { return manager.RedisConnected() && manager.Generation() == 2 })
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("manager did not stop")
+	}
+}
+
 func TestRedisCanBeDisabled(t *testing.T) {
 	store := newFakeManagerStore(3)
 	manager := newTestManager(store, slog.New(slog.NewTextHandler(&bytes.Buffer{}, nil)), 10*time.Millisecond)