fix: Stop the rollout watch when sign-in is refused

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
 			}
 

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)
+			}
+		})
+	}
+}