Merge pull request #181 from nlnwa/refactor/error-handling

Consistent error handling between subcommands

Author: maeb

arcreader/arcreader.go

@@ -17,7 +17,7 @@ type ArcFileReader struct {
 }
 
 func NewArcFileReader(fs afero.Fs, filename string, offset int64, opts ...gowarc.WarcRecordOption) (*ArcFileReader, error) {
-	file, err := fs.Open(filename) // For read access.
+	file, err := fs.Open(filename)
 	if err != nil {
 		return nil, err
 	}

arcreader/arcreader_test.go

@@ -1,7 +1,6 @@
 package arcreader
 
 import (
-	"context"
 	"path/filepath"
 	"testing"
 
@@ -43,9 +42,9 @@ func TestArcReader(t *testing.T) {
 				t.Fatalf("unexpected error: %v", err)
 			}
 
-			for record := range warc.NewIterator(context.Background(), arcFileReader, nil, 0, 0) {
-				if record.Err != nil {
-					t.Errorf("unexpected error: %v", record.Err)
+			for _, err := range warc.Records(arcFileReader, nil, 0, 0) {
+				if err != nil {
+					t.Errorf("unexpected error: %v", err)
 				}
 			}
 		})

arcreader/unmarshaler.go

@@ -246,7 +246,7 @@ func (u *unmarshaler) parseRecord(r *bufio.Reader, l1 string) (gowarc.WarcRecord
 func (u *unmarshaler) parseUrlRecordV1(l string) (gowarc.RecordType, string, string, time.Time, string, int64, error) {
 	reg := regexp.MustCompile(`([^ ]*) ([^ ]*) (\d*) ([^ ]*) (\d*)`)
 	subs := reg.FindStringSubmatch(l)
-	if subs == nil || len(subs) < 4 {
+	if len(subs) < 4 {
 		return 0, "", "", time.Time{}, "", 0, fmt.Errorf("could not parse ARC record from: %s", l)
 	}
 	url := subs[1]

cmd/aart/aart.go

@@ -2,7 +2,6 @@ package aart
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"fmt"
 	"image"
@@ -11,7 +10,6 @@ import (
 	_ "image/jpeg"
 	_ "image/png"
 	"io"
-	"os"
 
 	"github.com/nfnt/resize"
 	"github.com/nlnwa/gowarc/v2"
@@ -72,7 +70,7 @@ func NewCmdAart() *cobra.Command {
 	flags := AartFlags{}
 
 	cmd := &cobra.Command{
-		Use:    "aart",
+		Use:    "aart FILE",
 		Short:  "Show images",
 		Long:   ``,
 		Hidden: true,
@@ -123,30 +121,44 @@ func (o *AartOptions) Run() error {
 		return fmt.Errorf("failed to create WARC reader: %v", err)
 	}
 
-	for record := range warc.NewIterator(context.TODO(), wf, o.filter, o.recordNum, o.recordCount) {
-		if record.Err != nil {
-			return fmt.Errorf("failed reading records: %w", record.Err)
+	for record, err := range warc.Records(wf, o.filter, o.recordNum, o.recordCount) {
+		if err != nil {
+			return err
 		}
-		wr := record.WarcRecord
-
-		if b, ok := wr.Block().(gowarc.HttpResponseBlock); ok {
-			fmt.Printf("\u001B[2J\u001B[HUrl: %s\n\n", wr.WarcHeader().Get(gowarc.WarcTargetURI))
-			r, err := b.PayloadBytes()
-			if err != nil {
-				fmt.Fprintf(os.Stderr, "Failed to get payload bytes: %v\n", err)
-			}
-			err = display(r, o.width)
-			if err != nil {
-				fmt.Fprintf(os.Stderr, "Failed to display: %v\n", err)
-				continue
-			}
-			fmt.Printf("Hit enter to continue\n")
-			_, _ = fmt.Scanln()
+		err = o.handleRecord(record)
+		if err != nil {
+			return err
 		}
 	}
 	return nil
 }
 
+const ansiClearScreenEscapeSequence = "\u001B[2J\u001B[H"
+
+func (o *AartOptions) handleRecord(record warc.Record) error {
+	wr := record.WarcRecord
+
+	block, ok := wr.Block().(gowarc.HttpResponseBlock)
+	if !ok {
+		return nil
+	}
+
+	fmt.Print(ansiClearScreenEscapeSequence)
+	fmt.Printf("Url: %s\n\n", wr.WarcHeader().Get(gowarc.WarcTargetURI))
+	b, err := block.PayloadBytes()
+	if err != nil {
+		return fmt.Errorf("failed to get payload bytes: %w", err)
+	}
+	err = display(b, o.width)
+	if err != nil {
+		return fmt.Errorf("failed to display: %w", err)
+	}
+	fmt.Printf("Hit enter to continue\n")
+	_, _ = fmt.Scanln()
+
+	return nil
+}
+
 var asciiChar = "MND8OZ$7I?+=~:,.."
 
 func asciiArt(img image.Image, w, h int) []byte {

cmd/cat/cat.go

@@ -13,7 +13,6 @@ import (
 
 	"github.com/nlnwa/gowarc/v2"
 	"github.com/nlnwa/warchaeology/v3/cmd/internal/flag"
-	"github.com/nlnwa/warchaeology/v3/cmd/internal/log"
 	"github.com/nlnwa/warchaeology/v3/internal/filewalker"
 	"github.com/nlnwa/warchaeology/v3/internal/filter"
 	"github.com/nlnwa/warchaeology/v3/internal/warc"
@@ -34,14 +33,20 @@ const (
 	ShowPayload      = "payload"
 	ShowPayloadShort = "P"
 	ShowPayloadHelp  = "show payload"
+
+	Compression      = "compress"
+	CompressionShort = "z"
+	CompressionHelp  = "compress output (per record)"
 )
 
 type CatOptions struct {
 	paths             []string
 	offset            int64
 	recordNum         int
 	recordCount       int
+	force             bool
 	compress          bool
+	continueOnError   bool
 	filter            *filter.RecordFilter
 	writer            *writer
 	fileWalker        *filewalker.FileWalker
@@ -53,6 +58,7 @@ type CatFlags struct {
 	FilterFlags           flag.FilterFlags
 	WarcIteratorFlags     flag.WarcIteratorFlags
 	WarcRecordOptionFlags flag.WarcRecordOptionFlags
+	ErrorFlags            flag.ErrorFlags
 }
 
 func (f CatFlags) AddFlags(cmd *cobra.Command) {
@@ -61,11 +67,12 @@ func (f CatFlags) AddFlags(cmd *cobra.Command) {
 	f.FilterFlags.AddFlags(cmd)
 	f.WarcIteratorFlags.AddFlags(cmd)
 	f.WarcRecordOptionFlags.AddFlags(cmd)
+	f.ErrorFlags.AddFlags(cmd)
 
 	flags.BoolP(ShowWarcHeader, ShowWarcHeaderShort, false, ShowWarcHeaderHelp)
 	flags.BoolP(ShowProtocolHeader, ShowProtocolHeaderShort, false, ShowProtocolHeaderHelp)
 	flags.BoolP(ShowPayload, ShowPayloadShort, false, ShowPayloadHelp)
-	flags.BoolP("compress", "z", false, "output is compressed (per record)")
+	flags.BoolP(Compression, CompressionShort, false, CompressionHelp)
 }
 
 func (f CatFlags) ShowWarcHeader() bool {
@@ -108,11 +115,13 @@ func (f CatFlags) ToOptions() (*CatOptions, error) {
 
 	return &CatOptions{
 		paths:             fileList,
-		fileWalker:        fileWalker,
-		filter:            filter,
 		offset:            f.WarcIteratorFlags.Offset(),
 		recordCount:       f.WarcIteratorFlags.Limit(),
 		recordNum:         f.WarcIteratorFlags.RecordNum(),
+		force:             f.WarcIteratorFlags.Force(),
+		filter:            filter,
+		continueOnError:   f.ErrorFlags.ContinueOnError(),
+		fileWalker:        fileWalker,
 		compress:          f.Compress(),
 		writer:            writer,
 		warcRecordOptions: f.WarcRecordOptionFlags.ToWarcRecordOptions(),
@@ -126,10 +135,11 @@ func NewCmdCat() *cobra.Command {
 		Use:   "cat FILE/DIR ...",
 		Short: "Concatenate and print warc files",
 		Long:  ``,
-		Example: `Print all content from a WARC file
+		Example: `
+# Print all content from a WARC file (in principle the same as zcat)
 warc cat file1.warc.gz
 
-# Pipe payload from record #4 into the image viewer feh
+# Pipe the payload of the 4th record into the image viewer feh
 warc cat -n4 -P file1.warc.gz | feh -`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			o, err := flags.ToOptions()
@@ -145,7 +155,11 @@ warc cat -n4 -P file1.warc.gz | feh -`,
 				return err
 			}
 			cmd.SilenceUsage = true
-			return o.Run()
+			err = o.Run()
+			if errors.Is(err, context.Canceled) {
+				os.Exit(1)
+			}
+			return err
 		},
 	}
 
@@ -181,33 +195,36 @@ func (o *CatOptions) Run() error {
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
 	defer cancel()
 
-	closer, err := log.InitLogger(os.Stderr)
-	if err != nil {
-		return err
-	}
-	defer closer.Close()
+	for _, path := range o.paths {
+		err := o.fileWalker.Walk(ctx, path, func(fs afero.Fs, path string, err error) error {
+			if err != nil {
+				return err
+			}
 
-	walkFn := func(fs afero.Fs, path string, err error) error {
-		if err != nil {
-			return err
-		}
-		if err := o.catFile(ctx, fs, path); err != nil {
-			slog.Error(err.Error(), "path", path)
-		}
-		return nil
-	}
+			err = o.handleFile(ctx, fs, path)
+			if err != nil {
+				if !o.continueOnError {
+					cancel()
+				}
+				var recordErr warc.RecordError
+				if errors.As(err, &recordErr) {
+					slog.Error(recordErr.Error(), "path", path, "offset", recordErr.Offset())
+				} else {
+					slog.Error(err.Error(), "path", path)
+				}
+			}
 
-	for _, file := range o.paths {
-		err := o.fileWalker.Walk(file, walkFn)
+			return nil
+		})
 		if err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-// catFile reads a WARC file and writes the content to stdout
-func (o *CatOptions) catFile(ctx context.Context, fs afero.Fs, path string) error {
+// handleFile reads a WARC file and writes the content to stdout
+func (o *CatOptions) handleFile(ctx context.Context, fs afero.Fs, path string) error {
 	f, err := fs.Open(path)
 	if err != nil {
 		return err
@@ -216,24 +233,36 @@ func (o *CatOptions) catFile(ctx context.Context, fs afero.Fs, path string) erro
 	if err != nil {
 		return err
 	}
-	defer func() { _ = warcFileReader.Close() }()
+	defer func() {
+		_ = warcFileReader.Close()
+	}()
 
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
+	var lastOffset int64 = -1
 
-	for record := range warc.NewIterator(ctx, warcFileReader, o.filter, o.recordNum, o.recordCount) {
+	for record, err := range warc.Records(warcFileReader, o.filter, o.recordNum, o.recordCount) {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+		if err != nil {
+			// When forcing, avoid infinite loop by ensuring the iterator moves forward
+			if o.force && lastOffset != record.Offset {
+				slog.Warn(err.Error(), "offset", record.Offset, "path", path)
+				lastOffset = record.Offset
+				continue
+			}
+			return warc.Error(record, err)
+		}
 		if err := o.handleRecord(record); err != nil {
-			return err
+			return warc.Error(record, err)
 		}
 	}
 	return nil
 }
 
 func (o *CatOptions) handleRecord(record warc.Record) error {
 	defer record.Close()
-	if record.Err != nil {
-		return record.Err
-	}
 	var w io.Writer
 	if o.compress {
 		gw := gzip.NewWriter(os.Stdout)

cmd/cat/cat_test.go

@@ -3,13 +3,12 @@ package cat
 import (
 	"bytes"
 	"compress/gzip"
-	"context"
+	"errors"
 	"io"
 	"math"
 	"os"
 	"path/filepath"
 	"testing"
-	"time"
 
 	"github.com/nlnwa/gowarc/v2"
 	"github.com/nlnwa/warchaeology/v3/internal/warc"
@@ -25,15 +24,31 @@ var testFiles = map[string]string{
 	"samsung-with-error": filepath.Join(testDataDir, "warc", "samsung-with-error", "rec-33318048d933-20240317162652059-0.warc.gz"),
 }
 
+var tests = []struct {
+	name string
+	err  error
+}{
+	{
+		name: "empty",
+	},
+	{
+		name: "single-record",
+	},
+	{
+		name: "samsung-with-error",
+		err:  io.ErrUnexpectedEOF,
+	},
+}
+
 // TestWriteWarcRecord tests the writeWarcRecord function.
 func TestWriteWarcRecord(t *testing.T) {
-	for name, testFile := range testFiles {
-		t.Run(name, func(t *testing.T) {
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
 			// capture testFile variable from outer scope
-			testFile := testFile
+			test := test
 
 			// resolve test file path
-			testFile, err := filepath.Abs(testFile)
+			testFile, err := filepath.Abs(testFiles[test.name])
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -114,8 +129,11 @@ func TestWriteWarcRecord(t *testing.T) {
 			got := new(bytes.Buffer)
 			var currentOffset int64
 
-			for record := range warc.NewIterator(context.Background(), warcFileReader, nil, 0, 0) {
-				if record.Err != nil {
+			for record, err := range warc.Records(warcFileReader, nil, 0, 0) {
+				if err != nil {
+					if !errors.Is(err, test.err) {
+						t.Fatal(err)
+					}
 					break
 				}
 
@@ -155,11 +173,3 @@ want is %d bytes, got is %d bytes
 		})
 	}
 }
-
-func BenchmarkDummy(b *testing.B) {
-	// This is a dummy test, it should be replaced with something more
-	// meaningful in a later commit
-	for i := 0; i < b.N; i++ {
-		time.Sleep(1 * time.Nanosecond)
-	}
-}