From 9992e76a2b2d7fd7e9970b2d5487fbbac86c6948 Mon Sep 17 00:00:00 2001
From: Scot Schuchert-Wells <wells.scot@gmail.com>
Date: Wed, 11 Nov 2026 14:58:36 +0000
Subject: [PATCH] fix: Stop the rollout watch when sign-in is refused

The rollout watch treated every List error as transient and polled every
2 seconds forever. Once the sign-in token expired, each poll was refused
as unauthenticated, nothing was printed, and the process never exited.

Key changes:
- stop the watch on 401 or 403 and tell the user to run 'datumctl login'
- add a test where the credentials are refused partway through a watch
---
 internal/cmd/compute/watch/watch.go      | 13 +++--
 internal/cmd/compute/watch/watch_test.go | 60 ++++++++++++++++++++++++
 2 files changed, 70 insertions(+), 3 deletions(-)

diff --git a/internal/cmd/compute/watch/watch.go b/internal/cmd/compute/watch/watch.go
index f1d55b7d..eef0584d 100644
--- a/internal/cmd/compute/watch/watch.go
+++ b/internal/cmd/compute/watch/watch.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"time"
 
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -24,6 +25,8 @@ const (
 	phaseBlocked  deploymentPhase = "Blocked"
 )
 
+var pollInterval = 2 * time.Second
+
 type deploymentState struct {
 	placement    string
 	location     string
@@ -36,7 +39,8 @@ type deploymentState struct {
 
 // Rollout polls WorkloadDeployment objects for the given workload UID, printing
 // per-location progress rows as state changes. It returns when all deployments
-// reach Done, or when ctx is cancelled (Ctrl-C detach).
+// reach Done, when ctx is cancelled (Ctrl-C detach), or when the platform
+// refuses the credentials.
 func Rollout(ctx context.Context, c client.Client, out io.Writer, project string, workloadUID types.UID) error {
 	start := time.Now()
 
@@ -48,7 +52,7 @@ func Rollout(ctx context.Context, c client.Client, out io.Writer, project string
 	headerPrinted := false
 	states := map[string]*deploymentState{}
 
-	ticker := time.NewTicker(2 * time.Second)
+	ticker := time.NewTicker(pollInterval)
 	defer ticker.Stop()
 
 	for {
@@ -67,7 +71,10 @@ func Rollout(ctx context.Context, c client.Client, out io.Writer, project string
 				if ctx.Err() != nil {
 					return nil
 				}
-				// Transient error — keep polling.
+				if k8serrors.IsUnauthorized(err) || k8serrors.IsForbidden(err) {
+					_ = tw.Flush()
+					return fmt.Errorf("stopped watching the rollout, which continues in the background: %w — run 'datumctl login' to refresh your credentials", err)
+				}
 				continue
 			}
 
diff --git a/internal/cmd/compute/watch/watch_test.go b/internal/cmd/compute/watch/watch_test.go
index b90a4c48..e8840c2b 100644
--- a/internal/cmd/compute/watch/watch_test.go
+++ b/internal/cmd/compute/watch/watch_test.go
@@ -3,9 +3,21 @@
 package watch
 
 import (
+	"bytes"
+	"context"
+	"errors"
+	"strings"
 	"testing"
 	"time"
 
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
 	computev1alpha "go.datum.net/compute/api/v1alpha"
 )
 
@@ -179,3 +191,51 @@ func TestUpdateDeploymentState(t *testing.T) {
 		}
 	})
 }
+
+func TestRolloutStopsWhenCredentialsAreRefused(t *testing.T) {
+	pollInterval = 10 * time.Millisecond
+	t.Cleanup(func() { pollInterval = 2 * time.Second })
+
+	scheme := runtime.NewScheme()
+	if err := computev1alpha.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+
+	tests := []struct {
+		name   string
+		refuse error
+	}{
+		{"token expired", k8serrors.NewUnauthorized("jwt token is not active")},
+		{"access revoked", k8serrors.NewForbidden(schema.GroupResource{Resource: "workloaddeployments"}, "", errors.New("denied"))},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var lists int
+			c := fake.NewClientBuilder().WithScheme(scheme).WithInterceptorFuncs(interceptor.Funcs{
+				List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
+					lists++
+					if lists == 1 {
+						return c.List(ctx, list, opts...)
+					}
+					return tc.refuse
+				},
+			}).Build()
+
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+
+			var out bytes.Buffer
+			err := Rollout(ctx, c, &out, "p", types.UID("uid"))
+			if ctx.Err() != nil {
+				t.Fatal("watch kept polling after the credentials were refused")
+			}
+			if err == nil || !strings.Contains(err.Error(), "datumctl login") {
+				t.Fatalf("Rollout() error = %v, want one telling the user to sign in again", err)
+			}
+			if lists != 2 {
+				t.Errorf("List called %d times, want 2", lists)
+			}
+		})
+	}
+}