fix(eks): refuse an unreadable ownership record instead of treating it as absent

pkg/cli/clusterapi/distconfig.go

@@ -315,10 +315,16 @@ func boundEKSRegionFromConfig(name, configPath string) (string, bool, error) {
 // file's region would aim a destructive action at a cluster nobody named.
 func confirmConfigMatchesOwnership(name, configPath, configRegion string) error {
 	ownerships, err := state.ListEKSOwnershipStates(name)
-	if err != nil {
+	if errors.Is(err, state.ErrEKSOwnershipStateNotFound) {
 		// No usable record: the config is the only binding evidence there is, and it is already
 		// validated above. A cluster created before ownership records existed still works.
-		return nil //nolint:nilerr // absence of a record is not an error; the config binds instead.
+		return nil
+	}
+
+	if err != nil {
+		// A record that exists but cannot be read is not absence. Accepting the config here would
+		// let a stale rendered file bind unopposed, which is the redirect this check prevents.
+		return unreadableOwnershipError(name, err)
 	}
 
 	if len(ownerships) > 1 {
@@ -345,6 +351,19 @@ func confirmConfigMatchesOwnership(name, configPath, configRegion string) error
 	)
 }
 
+// unreadableOwnershipError refuses a lifecycle action whose ownership record exists but cannot be
+// read or parsed. err names the affected files.
+func unreadableOwnershipError(name string, err error) error {
+	return fmt.Errorf(
+		"%w: cluster %q has an EKS ownership record KSail cannot read, so it cannot confirm which"+
+			" region the cluster was created in; restore read access to the file or remove it and"+
+			" run `ksail cluster eks-bind` to record the region again: %w",
+		api.ErrInvalid,
+		name,
+		err,
+	)
+}
+
 // multiRegionOwnershipError reports same-named clusters recorded in several regions.
 func multiRegionOwnershipError(name string, ownerships []*state.EKSOwnershipState) error {
 	regions := make([]string, 0, len(ownerships))
@@ -380,6 +399,10 @@ func multiRegionOwnershipError(name string, ownerships []*state.EKSOwnershipStat
 // this refuses and lists them.
 func bindFromOwnershipRecord(name string) (*clusterprovisioner.EKSConfig, error) {
 	ownerships, err := state.ListEKSOwnershipStates(name)
+	if errors.Is(err, state.ErrEKSOwnershipStateUnreadable) {
+		return nil, unreadableOwnershipError(name, err)
+	}
+
 	if err != nil {
 		return nil, fmt.Errorf(
 			"%w: cluster %q has local KSail state but no eks config and no ownership record to bind"+

pkg/cli/clusterapi/local_service.go

@@ -523,22 +523,24 @@ func (s *Service) useDefaultClients() {
 }
 
 // resolveCluster finds the distribution and provider of an existing cluster, checking live
-// providers first and then the job store (for clusters still being provisioned).
+// providers first and then the job store (for clusters still being provisioned). An ownership record
+// that exists but cannot be read is returned as an error rather than as "not found", so the refusal
+// names the damaged file instead of claiming the cluster does not exist.
 func (s *Service) resolveCluster(
 	ctx context.Context,
 	name string,
-) (v1alpha1.Distribution, v1alpha1.Provider, bool) {
+) (v1alpha1.Distribution, v1alpha1.Provider, bool, error) {
 	live := s.enumerate(ctx)
 	if cluster, ok := live[name]; ok {
-		return cluster.Distribution, cluster.Provider, true
+		return cluster.Distribution, cluster.Provider, true, nil
 	}
 
 	s.mu.Lock()
 
 	if current, ok := s.jobs[name]; ok {
 		s.mu.Unlock()
 
-		return current.distribution, current.provider, true
+		return current.distribution, current.provider, true, nil
 	}
 
 	s.mu.Unlock()
@@ -555,10 +557,14 @@ func (s *Service) resolveCluster(
 	// account for.
 	_, ownershipErr := state.ListEKSOwnershipStates(name)
 	if ownershipErr == nil {
-		return v1alpha1.DistributionEKS, v1alpha1.ProviderAWS, true
+		return v1alpha1.DistributionEKS, v1alpha1.ProviderAWS, true, nil
 	}
 
-	return "", "", false
+	if errors.Is(ownershipErr, state.ErrEKSOwnershipStateUnreadable) {
+		return "", "", false, unreadableOwnershipError(name, ownershipErr)
+	}
+
+	return "", "", false, nil
 }
 
 // dockerFactory adapts the Service's provisioner factory to the discovery DockerFactory shape,
@@ -580,7 +586,11 @@ func (s *Service) startJob(
 	name string,
 	phase v1alpha1.ClusterPhase,
 ) (v1alpha1.Spec, error) {
-	distribution, provider, ok := s.resolveCluster(ctx, name)
+	distribution, provider, ok, resolveErr := s.resolveCluster(ctx, name)
+	if resolveErr != nil {
+		return v1alpha1.Spec{}, resolveErr
+	}
+
 	if !ok {
 		return v1alpha1.Spec{}, fmt.Errorf("%w: %q", api.ErrNotFound, name)
 	}

pkg/cli/clusterapi/local_service_test.go

@@ -2429,6 +2429,42 @@ func TestDeleteResolvesAPersistedEKSTargetOutsideTheSelectedRegion(t *testing.T)
 		"a cluster KSail holds an ownership record for is not missing; the selected region is")
 }
 
+// TestLifecycleReportsAnUnreadableOwnershipRecordInsteadOfNotFound covers a cluster reachable only
+// through its ownership record when that record cannot be read. The mutation must still be refused,
+// but as the unreadable record — naming the file — rather than as a cluster that does not exist,
+// which would hide both the cause and its fix.
+func TestLifecycleReportsAnUnreadableOwnershipRecordInsteadOfNotFound(t *testing.T) {
+	home := t.TempDir()
+	t.Setenv("HOME", home)
+
+	const clusterName = "unreadable-other-region"
+
+	service := newTestService(map[v1alpha1.Distribution]*fakeProvisioner{
+		v1alpha1.DistributionEKS: {},
+	})
+
+	recordDir := filepath.Join(home, ".ksail", "clusters", clusterName)
+	require.NoError(t, os.MkdirAll(recordDir, 0o750))
+
+	recordPath := filepath.Join(recordDir, "eks-ownership-ap-southeast-2.json")
+	require.NoError(t, os.WriteFile(recordPath, []byte(`{"version":1,"region":"ap-so`), 0o600))
+
+	ctx := context.Background()
+
+	for action, run := range map[string]func() error{
+		"delete": func() error { return service.Delete(ctx, "default", clusterName) },
+		"start":  func() error { return service.Start(ctx, "default", clusterName) },
+		"stop":   func() error { return service.Stop(ctx, "default", clusterName) },
+	} {
+		err := run()
+		require.ErrorIs(t, err, state.ErrEKSOwnershipStateUnreadable, action)
+		require.ErrorIs(t, err, api.ErrInvalid, action)
+		require.NotErrorIs(t, err, api.ErrNotFound, action)
+		assert.Contains(t, err.Error(), recordPath, action)
+		assert.Contains(t, err.Error(), "ksail cluster eks-bind", action)
+	}
+}
+
 // writeStateEKSConfig puts an eks.yaml under ~/.ksail/clusters/<name> carrying region, standing in
 // for a file an earlier KSail wrote from whatever region happened to be selected at the time.
 func writeStateEKSConfig(t *testing.T, name, region string) {
@@ -2543,6 +2579,50 @@ func TestBoundEKSConfigAcceptsAConfigWithNoOwnershipRecord(t *testing.T) {
 	assert.Equal(t, "eu-north-1", region, "the config alone still binds when no record exists")
 }
 
+// TestBoundEKSConfigRefusesAConfigBesideATruncatedOwnershipRecord closes the corruption route to the
+// stale-config redirect. A record that exists but does not parse used to read as "no record", so the
+// config-only path above accepted a rendered file the record might contradict. It must refuse and
+// name the damaged file instead.
+func TestBoundEKSConfigRefusesAConfigBesideATruncatedOwnershipRecord(t *testing.T) {
+	home := t.TempDir()
+	t.Setenv("HOME", home)
+	t.Setenv("AWS_REGION", "us-west-2")
+
+	const name = "truncated-record"
+
+	saveEKSClusterSpec(t, name)
+	writeStateEKSConfig(t, name, "us-west-2")
+
+	recordPath := filepath.Join(home, ".ksail", "clusters", name, "eks-ownership-eu-north-1.json")
+	require.NoError(t, os.WriteFile(recordPath, []byte(`{"version":1,"region":"eu-no`), 0o600))
+
+	_, _, err := clusterapi.ExportEKSConfigForCreate(name)
+	require.ErrorIs(t, err, api.ErrInvalid)
+	require.ErrorIs(t, err, state.ErrEKSOwnershipStateUnreadable)
+	assert.Contains(t, err.Error(), recordPath, "the error must name the file that cannot be read")
+}
+
+// TestBoundEKSConfigRefusesToBindFromAnUnreadableOwnershipRecord covers the no-config path: with no
+// eks.yaml, binding comes from the record alone, so an unreadable one must say so rather than report
+// that no record exists.
+func TestBoundEKSConfigRefusesToBindFromAnUnreadableOwnershipRecord(t *testing.T) {
+	home := t.TempDir()
+	t.Setenv("HOME", home)
+	t.Setenv("AWS_REGION", "us-west-2")
+
+	const name = "unreadable-record-no-config"
+
+	saveEKSClusterSpec(t, name)
+
+	recordPath := filepath.Join(home, ".ksail", "clusters", name, "eks-ownership-eu-north-1.json")
+	require.NoError(t, os.WriteFile(recordPath, []byte("{"), 0o600))
+
+	_, _, err := clusterapi.ExportEKSConfigForCreate(name)
+	require.ErrorIs(t, err, api.ErrInvalid)
+	require.ErrorIs(t, err, state.ErrEKSOwnershipStateUnreadable)
+	assert.Contains(t, err.Error(), "cannot read")
+}
+
 // TestCreateRefusesANameWhoseEKSCreateStateRemains covers the CREATE half of the region binding.
 //
 // The binding that makes delete/start/stop follow the creation region must never steer a create.