Allow explicit recovery after ZIP member errors

README.md

@@ -168,6 +168,13 @@ are terminal. The caller owns the input, and `Close` releases decoder resources
 without closing or draining it. A TAR end marker ends iteration, so trailing
 data and compression trailers beyond that marker may remain unchecked.
 
+After a ZIP member error, callers can explicitly call `stream.Recover()` to
+discard that member and continue with `Next`. The central directory provides
+the next member's offset. TAR-based formats, including conda, return
+`errors.ErrUnsupported` from `Recover`, as does a stream with no pending error.
+Cancellation and limit errors remain terminal. Recovery does not validate unread
+data in the discarded member.
+
 ZIP and conda require random access to their compressed input. `OpenStream`
 buffers that input within `MaxInputBytes`; `OpenStreamBytes(name, data, options)`
 reuses an existing byte slice without copying it. Keep that slice unchanged

stream.go

@@ -240,6 +240,32 @@ func (s *Stream) Read(p []byte) (int, error) {
 	return n, err
 }
 
+// Recover abandons a failed ZIP entry so the next call to Next can continue.
+// TAR-based formats, including conda, return errors.ErrUnsupported, as does a
+// stream with no pending error. Cancellation and resource-limit errors remain
+// terminal. Recovery does not validate the discarded body.
+func (s *Stream) Recover() error {
+	if s.closed {
+		return fs.ErrClosed
+	}
+	if err := s.ctx.Err(); err != nil {
+		return err
+	}
+	if s.err == nil || errors.Is(s.err, io.EOF) {
+		return errors.ErrUnsupported
+	}
+	if errors.Is(s.err, ErrInputLimit) || errors.Is(s.err, ErrEntryLimit) || errors.Is(s.err, ErrEntrySizeLimit) || errors.Is(s.err, ErrDecompressLimit) {
+		return s.err
+	}
+	zip, ok := s.source.(*zipStream)
+	if !ok {
+		return errors.ErrUnsupported
+	}
+	zip.recover()
+	s.err = nil
+	return nil
+}
+
 // Close releases resources without draining or closing the input. Unread
 // content and compressed trailers after a TAR end marker are not validated.
 func (s *Stream) Close() error {

stream_recover_test.go

@@ -0,0 +1,150 @@
+package archives
+
+import (
+	"archive/zip"
+	"bytes"
+	"compress/flate"
+	"context"
+	"errors"
+	"io"
+	"testing"
+)
+
+func corruptZipMember(t *testing.T, method uint16) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	w := zip.NewWriter(&buf)
+	for _, name := range []string{"broken", "valid"} {
+		member, err := w.CreateHeader(&zip.FileHeader{Name: name, Method: method})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := io.WriteString(member, name); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	data := buf.Bytes()
+	reader, err := zip.NewReader(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	offset, err := reader.File[0].DataOffset()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if method == zip.Store {
+		data[offset] ^= 1
+	} else {
+		data[offset] = 0x07 // Reserved DEFLATE block type.
+	}
+	return data
+}
+
+func TestStreamRecoverZIP(t *testing.T) {
+	for _, method := range []uint16{zip.Store, zip.Deflate} {
+		for _, skip := range []bool{false, true} {
+			assertRecoverZIP(t, method, skip)
+		}
+	}
+}
+
+func assertRecoverZIP(t *testing.T, method uint16, skip bool) {
+	t.Helper()
+	s, err := OpenStreamBytes("test.zip", corruptZipMember(t, method), StreamOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Next(); err != nil {
+		t.Fatal(err)
+	}
+	if skip {
+		_, err = s.Next()
+	} else {
+		_, err = io.ReadAll(s)
+	}
+	var corrupt flate.CorruptInputError
+	if (method == zip.Store && !errors.Is(err, zip.ErrChecksum)) || (method == zip.Deflate && !errors.As(err, &corrupt)) {
+		t.Fatalf("member error: %v", err)
+	}
+	if err := s.Recover(); err != nil {
+		t.Fatal(err)
+	}
+	next, err := s.Next()
+	if err != nil || next.Path != "valid" {
+		t.Fatalf("next: %+v %v", next, err)
+	}
+	body, err := io.ReadAll(s)
+	if err != nil || string(body) != "valid" {
+		t.Fatalf("body: %q %v", body, err)
+	}
+	if _, err := s.Next(); err != io.EOF {
+		t.Fatal(err)
+	}
+	if err := s.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestStreamRecoverPreservesLimitsAndCancellation(t *testing.T) {
+	for _, options := range []StreamOptions{{MaxEntryBytes: 1}, {MaxExpandedBytes: 1}} {
+		s, err := OpenStreamBytes("test.zip", corruptZipMember(t, zip.Store), options)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, original := s.Next()
+		if original == nil || s.Recover() != original {
+			t.Fatalf("limit bypassed: %v", original)
+		}
+		_ = s.Close()
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	s, err := OpenStreamBytesContext(ctx, "test.zip", corruptZipMember(t, zip.Store), StreamOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	if err := s.Recover(); !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	_ = s.Close()
+}
+
+func TestStreamRecoverUnsupported(t *testing.T) {
+	raw := tarWithPayloads(t, make([]byte, 1024))[:1023]
+	for _, name := range []string{"test.tar", "test.tar.gz", "test.conda"} {
+		s, err := OpenStreamBytes(name, streamTestArchive(t, name, raw), StreamOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := s.Next(); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := io.ReadAll(s); !errors.Is(err, io.ErrUnexpectedEOF) {
+			t.Fatalf("%s: %v", name, err)
+		}
+		if err := s.Recover(); !errors.Is(err, errors.ErrUnsupported) {
+			t.Fatalf("%s: %v", name, err)
+		}
+		_ = s.Close()
+	}
+}
+
+func TestStreamRecoverWithoutError(t *testing.T) {
+	s, err := OpenStreamBytes("test.zip", createTestZip(), StreamOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = s.Close() }()
+	if err := s.Recover(); !errors.Is(err, errors.ErrUnsupported) {
+		t.Fatal(err)
+	}
+	if err := consumeStream(s, true); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Recover(); !errors.Is(err, errors.ErrUnsupported) {
+		t.Fatal(err)
+	}
+}

stream_zip.go

@@ -85,6 +85,15 @@ func (z *zipStream) Close() error {
 	return nil
 }
 
+func (z *zipStream) recover() {
+	if z.current != nil {
+		// A decompressor can repeat the already-reported member error on Close.
+		_ = z.current.Close()
+	}
+	z.current = nil
+	z.file = nil
+}
+
 type condaStream struct {
 	outer *Stream
 	inner *tarStream