[bug] - correclty handle nested archived directories (#1778)

pkg/handlers/archive.go

@@ -78,7 +78,7 @@ func (a *Archive) FromFile(originalCtx context.Context, data io.Reader) chan []b
 			if errors.Is(err, archiver.ErrNoMatch) {
 				return
 			}
-			logger.V(2).Info("Error unarchiving chunk.")
+			logger.Error(err, "error unarchiving chunk.")
 		}
 	}()
 	return archiveChan

pkg/handlers/archive_test.go

@@ -9,6 +9,7 @@ import (
 	"regexp"
 	"strings"
 	"testing"
+	"time"
 
 	diskbufferreader "github.com/bill-rich/disk-buffer-reader"
 	"github.com/stretchr/testify/assert"
@@ -248,3 +249,25 @@ func TestOpenInvalidArchive(t *testing.T) {
 	err := a.openArchive(ctx, 0, reader, archiveChan)
 	assert.Error(t, err)
 }
+
+func TestNestedDirArchive(t *testing.T) {
+	file, err := os.Open("testdata/dir-archive.zip")
+	assert.Nil(t, err)
+	defer file.Close()
+
+	ctx, cancel := logContext.WithTimeout(logContext.Background(), 5*time.Second)
+	defer cancel()
+	sourceChan := make(chan *sources.Chunk, 1)
+
+	go func() {
+		defer close(sourceChan)
+		HandleFile(ctx, file, &sources.Chunk{}, sourceChan)
+	}()
+
+	count := 0
+	want := 4
+	for range sourceChan {
+		count++
+	}
+	assert.Equal(t, want, count)
+}

pkg/handlers/handlers.go

@@ -47,42 +47,44 @@ func HandleFile(ctx context.Context, file io.Reader, chunkSkel *sources.Chunk, c
 		// an io.MultiReader, which is used by the SpecializedHandler.
 		reReader, err := diskbufferreader.New(file)
 		if err != nil {
-			aCtx.Logger().Error(err, "error creating re-reader reader")
+			aCtx.Logger().Error(err, "error creating reusable reader")
 			return false
 		}
-		defer reReader.Close()
 
-		// Check if the handler implements SpecializedHandler and process accordingly.
-		if specialHandler, ok := h.(SpecializedHandler); ok {
-			file, isSpecial, err := specialHandler.HandleSpecialized(aCtx, reReader)
-			if isSpecial {
-				return handleChunks(aCtx, h.FromFile(ctx, file), chunkSkel, chunksChan)
-			}
-
-			if err != nil {
-				aCtx.Logger().Error(err, "error handling file")
-			}
+		if success := processHandler(aCtx, h, reReader, chunkSkel, chunksChan); success {
+			return true
 		}
-
-		if err := reReader.Reset(); err != nil {
-			aCtx.Logger().Error(err, "error resetting re-reader")
-			return false
-		}
-		if _, isType := h.IsFiletype(aCtx, reReader); !isType {
-			continue
-		}
-
-		if err := reReader.Reset(); err != nil {
-			aCtx.Logger().Error(err, "error resetting re-reader")
-			return false
-		}
-		reReader.Stop()
-		return handleChunks(aCtx, h.FromFile(ctx, reReader), chunkSkel, chunksChan)
 	}
 
 	return false
 }
 
+func processHandler(ctx logContext.Context, h Handler, reReader *diskbufferreader.DiskBufferReader, chunkSkel *sources.Chunk, chunksChan chan *sources.Chunk) bool {
+	defer reReader.Close()
+	defer reReader.Stop()
+
+	if specialHandler, ok := h.(SpecializedHandler); ok {
+		file, isSpecial, err := specialHandler.HandleSpecialized(ctx, reReader)
+		if isSpecial {
+			return handleChunks(ctx, h.FromFile(ctx, file), chunkSkel, chunksChan)
+		}
+		if err != nil {
+			ctx.Logger().Error(err, "error handling file")
+		}
+	}
+
+	if _, err := reReader.Seek(0, io.SeekStart); err != nil {
+		ctx.Logger().Error(err, "error seeking to start of file")
+		return false
+	}
+
+	if _, isType := h.IsFiletype(ctx, reReader); !isType {
+		return false
+	}
+
+	return handleChunks(ctx, h.FromFile(ctx, reReader), chunkSkel, chunksChan)
+}
+
 func handleChunks(ctx context.Context, handlerChan chan []byte, chunkSkel *sources.Chunk, chunksChan chan *sources.Chunk) bool {
 	for {
 		select {