[feat] - Introduce Fatal/Non-Fatal File Handling Errors

pkg/handlers/ar.go

@@ -42,14 +42,18 @@ func (h *arHandler) HandleFile(ctx logContext.Context, input fileReader) chan Da
 				} else {
 					panicErr = fmt.Errorf("panic occurred: %v", r)
 				}
-				ctx.Logger().Error(panicErr, "Panic occurred when attempting to open ar archive")
+				dataOrErrChan <- DataOrErr{
+					Err: fmt.Errorf("%w: panic error: %v", ErrProcessingFatal, panicErr),
+				}
 			}
 		}()
 
 		start := time.Now()
 		arReader, err := deb.LoadAr(input)
 		if err != nil {
-			ctx.Logger().Error(err, "Error loading AR file")
+			dataOrErrChan <- DataOrErr{
+				Err: fmt.Errorf("%w: loading AR error: %v", ErrProcessingFatal, err),
+			}
 			return
 		}
 
@@ -85,12 +89,19 @@ func (h *arHandler) processARFiles(ctx logContext.Context, reader *deb.Ar, dataO
 
 			rdr, err := newMimeTypeReader(arEntry.Data)
 			if err != nil {
-				return fmt.Errorf("error creating mime-type reader: %w", err)
+				dataOrErrChan <- DataOrErr{
+					Err: fmt.Errorf("%w: error creating AR mime-type reader: %v", ErrProcessingWarning, err),
+				}
+				h.metrics.incErrors()
+				continue
 			}
 
 			if err := h.handleNonArchiveContent(fileCtx, rdr, dataOrErrChan); err != nil {
-				fileCtx.Logger().Error(err, "error handling archive content in AR")
+				dataOrErrChan <- DataOrErr{
+					Err: fmt.Errorf("%w: error handling archive content in AR: %v", ErrProcessingWarning, err),
+				}
 				h.metrics.incErrors()
+				continue
 			}
 
 			h.metrics.incFilesProcessed()

pkg/handlers/archive.go

@@ -66,7 +66,9 @@ func (h *archiveHandler) HandleFile(ctx logContext.Context, input fileReader) ch
 				} else {
 					panicErr = fmt.Errorf("panic occurred: %v", r)
 				}
-				ctx.Logger().Error(panicErr, "Panic occurred when attempting to open archive")
+				dataOrErrChan <- DataOrErr{
+					Err: fmt.Errorf("%w: panic error: %v", ErrProcessingFatal, panicErr),
+				}
 			}
 		}()
 

pkg/handlers/default.go

@@ -106,9 +106,9 @@ func (h *defaultHandler) handleNonArchiveContent(
 		dataOrErr := DataOrErr{}
 		if err := data.Error(); err != nil {
 			h.metrics.incErrors()
-			dataOrErr.Err = fmt.Errorf("%w: error reading chunk", err)
+			dataOrErr.Err = fmt.Errorf("%w: error reading chunk: %v", ErrProcessingWarning, err)
 			if writeErr := common.CancellableWrite(ctx, dataOrErrChan, dataOrErr); writeErr != nil {
-				return fmt.Errorf("%w: error writing to data channel", writeErr)
+				return fmt.Errorf("%w: error writing to data channel: %v", ErrProcessingFatal, writeErr)
 			}
 			continue
 		}

pkg/handlers/handlers.go

@@ -2,6 +2,7 @@ package handlers
 
 import (
 	"bufio"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -38,7 +39,16 @@ type fileReader struct {
 	*iobuf.BufferedReadSeeker
 }
 
-var ErrEmptyReader = errors.New("reader is empty")
+var (
+	ErrEmptyReader = errors.New("reader is empty")
+
+	// ErrProcessingFatal indicates a severe error that requires stopping the file processing.
+	ErrProcessingFatal = errors.New("fatal error processing file")
+
+	// ErrProcessingWarning indicates a recoverable error that can be logged,
+	// allowing processing to continue.
+	ErrProcessingWarning = errors.New("error processing file")
+)
 
 // mimeTypeReader wraps an io.Reader with MIME type information.
 // This type is used to pass content through the processing pipeline
@@ -351,7 +361,7 @@ func HandleFile(
 // handleChunksWithError processes data and errors received from the dataErrChan channel.
 // For each DataOrErr received:
 // - If it contains data, the function creates a chunk based on chunkSkel and reports it through the reporter.
-// - If it contains an error, the function logs the error.
+// - If it contains an error, the function returns the error immediately.
 // The function also listens for context cancellation to gracefully terminate processing if the context is done.
 // It returns nil upon successful processing of all data, or the first encountered error.
 func handleChunksWithError(
@@ -369,7 +379,10 @@ func handleChunksWithError(
 				return nil
 			}
 			if dataOrErr.Err != nil {
-				ctx.Logger().Error(dataOrErr.Err, "error processing chunk")
+				if isFatal(dataOrErr.Err) {
+					return dataOrErr.Err
+				}
+				ctx.Logger().Error(dataOrErr.Err, "non-critical error processing chunk")
 				continue
 			}
 			if len(dataOrErr.Data) > 0 {
@@ -384,3 +397,21 @@ func handleChunksWithError(
 		}
 	}
 }
+
+// isFatal determines whether the given error is a fatal error that should
+// terminate processing the current file, or a non-critical error that can be logged and ignored.
+// "Fatal" errors include context cancellation, deadline exceeded, and the
+// ErrProcessingFatal error. Non-fatal errors include the ErrProcessingWarning
+// error as well as any other error that is not one of the fatal errors.
+func isFatal(err error) bool {
+	switch {
+	case errors.Is(err, context.Canceled) ||
+		errors.Is(err, context.DeadlineExceeded) ||
+		errors.Is(err, ErrProcessingFatal):
+		return true
+	case errors.Is(err, ErrProcessingWarning):
+		return false
+	default:
+		return false
+	}
+}

pkg/handlers/handlers_test.go

@@ -3,6 +3,7 @@ package handlers
 import (
 	"archive/zip"
 	"bytes"
+	stdctx "context"
 	"errors"
 	"fmt"
 	"io"
@@ -781,3 +782,79 @@ func getGitCommitHash(t *testing.T, gitDir string) string {
 	commitHash := strings.TrimSpace(string(hashBytes))
 	return commitHash
 }
+
+type mockReporter struct{ reportedChunks int }
+
+func (m *mockReporter) ChunkOk(context.Context, sources.Chunk) error {
+	m.reportedChunks++
+	return nil
+}
+
+func (m *mockReporter) ChunkErr(context.Context, error) error { return nil }
+
+func TestHandleChunksWithError(t *testing.T) {
+	tests := []struct {
+		name                   string
+		input                  []DataOrErr
+		expectedErr            error
+		expectedReportedChunks int
+	}{
+		{
+			name:  "Non-Critical Error",
+			input: []DataOrErr{{Err: ErrProcessingWarning}},
+		},
+		{
+			name:        "Critical Error",
+			input:       []DataOrErr{{Err: ErrProcessingFatal}},
+			expectedErr: ErrProcessingFatal,
+		},
+		{
+			name: "No Error",
+			input: []DataOrErr{
+				{Data: []byte("test data")},
+				{Data: []byte("more data")},
+			},
+			expectedReportedChunks: 2,
+		},
+		{
+			name:        "Context Canceled",
+			input:       []DataOrErr{{Err: stdctx.Canceled}},
+			expectedErr: stdctx.Canceled,
+		},
+		{
+			name:        "Context Deadline Exceeded",
+			input:       []DataOrErr{{Err: stdctx.DeadlineExceeded}},
+			expectedErr: stdctx.DeadlineExceeded,
+		},
+		{
+			name:  "EOF Error",
+			input: []DataOrErr{{Err: io.EOF}},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			ctx := context.Background()
+			chunkSkel := &sources.Chunk{}
+			reporter := new(mockReporter)
+
+			dataErrChan := make(chan DataOrErr, len(tc.input))
+			for _, de := range tc.input {
+				dataErrChan <- de
+			}
+			close(dataErrChan)
+
+			err := handleChunksWithError(ctx, dataErrChan, chunkSkel, reporter)
+
+			if tc.expectedErr != nil {
+				assert.ErrorIs(t, err, tc.expectedErr, "handleChunksWithError should return the expected error")
+			} else {
+				assert.NoError(t, err, "handleChunksWithError should not return an error for non-critical errors")
+			}
+
+			assert.Equal(t, tc.expectedReportedChunks, reporter.reportedChunks, "should have reported the expected number of chunks")
+		})
+	}
+}

pkg/handlers/rpm.go

@@ -42,20 +42,26 @@ func (h *rpmHandler) HandleFile(ctx logContext.Context, input fileReader) chan D
 				} else {
 					panicErr = fmt.Errorf("panic occurred: %v", r)
 				}
-				ctx.Logger().Error(panicErr, "Panic occurred when attempting to open rpm archive")
+				dataOrErrChan <- DataOrErr{
+					Err: fmt.Errorf("%w: panic error: %v", ErrProcessingFatal, panicErr),
+				}
 			}
 		}()
 
 		start := time.Now()
 		rpm, err := rpmutils.ReadRpm(input)
 		if err != nil {
-			ctx.Logger().Error(err, "Error reading rpm file")
+			dataOrErrChan <- DataOrErr{
+				Err: fmt.Errorf("%w: reading rpm error: %v", ErrProcessingFatal, err),
+			}
 			return
 		}
 
 		reader, err := rpm.PayloadReaderExtended()
 		if err != nil {
-			ctx.Logger().Error(err, "Error reading rpm file")
+			dataOrErrChan <- DataOrErr{
+				Err: fmt.Errorf("%w: uncompressing rpm error: %v", ErrProcessingFatal, err),
+			}
 			return
 		}
 
@@ -99,7 +105,9 @@ func (h *rpmHandler) processRPMFiles(
 			}
 
 			if err := h.handleNonArchiveContent(fileCtx, rdr, dataOrErrChan); err != nil {
-				fileCtx.Logger().Error(err, "error handling archive content in RPM")
+				dataOrErrChan <- DataOrErr{
+					Err: fmt.Errorf("%w: error processing RPM archive: %v", ErrProcessingWarning, err),
+				}
 				h.metrics.incErrors()
 			}