[Registry] Propagate Components CSV read errors

docs/agent-instructions/registration.md

@@ -39,6 +39,14 @@ Registration code is intentionally permissive on input shape:
 Do not tighten these behaviors without coordinating with every downstream importer
 (Meshery Server model import, mesheryctl, generators).
 
+## Component CSV ingestion
+
+`ComponentCSVHelper.ParseComponentsSheet` returns CSV parsing and read failures,
+including failures after the file opens. Rows read before a failure can already
+be present in `Components`, so callers must check the returned error before
+using that data to update the registry. Successful completion includes every
+selected row, including the final row when filtering by model.
+
 ## SVG mutation caveat
 
 Registration mutates asset references as part of ingestion: model and component SVG

registry/component.go

@@ -264,7 +264,8 @@ func (mch *ComponentCSVHelper) GetColumns() ([]string, error) {
 }
 
 func (mch *ComponentCSVHelper) ParseComponentsSheet(modelName string) error {
-	ch := make(chan ComponentCSV, 1)
+	// Each row must be consumed before the parser can signal completion.
+	ch := make(chan ComponentCSV)
 	errorChan := make(chan error, 1)
 	csvReader, err := csv.NewCSVParser[ComponentCSV](mch.CSVPath, rowIndex, nil, func(_ []string, _ []string) bool {
 		return true
@@ -275,6 +276,7 @@ func (mch *ComponentCSVHelper) ParseComponentsSheet(modelName string) error {
 	}
 
 	go func() {
+		defer close(errorChan)
 		Log.Info("Parsing Components...")
 
 		err := csvReader.Parse(ch, errorChan)
@@ -283,6 +285,7 @@ func (mch *ComponentCSVHelper) ParseComponentsSheet(modelName string) error {
 		}
 	}()
 
+	var parseErr error
 	for {
 		select {
 
@@ -298,11 +301,13 @@ func (mch *ComponentCSVHelper) ParseComponentsSheet(modelName string) error {
 			}
 			mch.Components[data.Registrant][data.Model] = append(mch.Components[data.Registrant][data.Model], data)
 			// Log.Info(fmt.Sprintf("Reading registrant [%s] model [%s] component [%s]", data.Registrant, data.Model, data.Component))
-		case err := <-errorChan:
-			Log.Error(err)
-
-		case <-csvReader.Context.Done():
-			return nil
+		case err, open := <-errorChan:
+			// Parse cancels its context before returning its error. Wait for
+			// the producer to finish forwarding errors before returning.
+			if !open {
+				return parseErr
+			}
+			parseErr = err
 		}
 	}
 }

registry/component_csv_test.go

@@ -0,0 +1,80 @@
+package registry
+
+import (
+	"io"
+	"path/filepath"
+	"reflect"
+	"testing"
+
+	meshkiterrors "github.com/meshery/meshkit/errors"
+	"github.com/meshery/meshkit/utils"
+)
+
+func TestParseComponentsSheet(t *testing.T) {
+	previousLog := Log
+	Log = SetupLogger("component-csv-test", false, io.Discard)
+	t.Cleanup(func() { Log = previousLog })
+
+	newHelper := func(t *testing.T, fixture string) *ComponentCSVHelper {
+		t.Helper()
+		path := filepath.Join("testdata", fixture)
+		helper, err := NewComponentCSVHelper("", "Components", 0, path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return helper
+	}
+
+	t.Run("returns a read error after parsing a component", func(t *testing.T) {
+		helper := newHelper(t, "components-read-error.csv")
+		err := helper.ParseComponentsSheet("")
+		if err == nil {
+			t.Fatal("expected the CSV read error after the valid component row")
+		}
+		if code := meshkiterrors.GetCode(err); code != utils.ErrReadFileCode {
+			t.Fatalf("expected read error code %q, got %q: %v", utils.ErrReadFileCode, code, err)
+		}
+		components := helper.Components["meshery"]["kubernetes"]
+		if len(components) != 1 || components[0].Component != "Pod" {
+			t.Fatalf("expected the component before the read failure, got %#v", components)
+		}
+	})
+
+	for _, test := range []struct {
+		name  string
+		model string
+		want  map[string][]string
+	}{
+		{
+			name: "retains every row on successful completion",
+			want: map[string][]string{
+				"meshery/kubernetes": {"Pod", "Deployment"},
+				"meshery/istio":      {"VirtualService"},
+			},
+		},
+		{
+			name:  "filters by model without dropping the last matching row",
+			model: "kubernetes",
+			want:  map[string][]string{"meshery/kubernetes": {"Pod", "Deployment"}},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			helper := newHelper(t, "components.csv")
+			if err := helper.ParseComponentsSheet(test.model); err != nil {
+				t.Fatal(err)
+			}
+			got := make(map[string][]string)
+			for registrant, models := range helper.Components {
+				for model, components := range models {
+					for _, component := range components {
+						key := registrant + "/" + model
+						got[key] = append(got[key], component.Component)
+					}
+				}
+			}
+			if !reflect.DeepEqual(got, test.want) {
+				t.Fatalf("parsed components = %#v, want %#v", got, test.want)
+			}
+		})
+	}
+}

registry/testdata/components-read-error.csv

@@ -0,0 +1,4 @@
+,,
+registrant,model,component
+meshery,kubernetes,Pod
+"unterminated

registry/testdata/components.csv

@@ -0,0 +1,5 @@
+,,
+registrant,model,component
+meshery,kubernetes,Pod
+meshery,istio,VirtualService
+meshery,kubernetes,Deployment