Allow skipping unread ZIP entries after size rejection

README.md

@@ -175,6 +175,14 @@ the next member's offset. TAR-based formats, including conda, return
 Cancellation and limit errors remain terminal. Recovery does not validate unread
 data in the discarded member.
 
+`stream.Skip()` abandons an unread ZIP entry without decompressing or validating
+its body. This also clears a size-limit rejection from `Next`, which returns the
+rejected entry alongside its error. The entry still counts toward `MaxEntries`,
+but its unread body does not consume the expanded-byte budget. Later entries can
+use the remaining budget. `Skip` rejects entries already read and does not support
+TAR-based formats or conda. Input limits, entry-count limits and cancellation
+remain terminal.
+
 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

@@ -45,6 +45,9 @@ type Stream struct {
 	unknownSize bool
 	entryBytes  int64
 	entryPath   string
+	pending     *StreamEntry
+	entryRead   bool
+	reserved    int64
 }
 
 type streamSource interface {
@@ -189,7 +192,8 @@ func streamCompression(format string) (string, bool) {
 // the body is exposed. Unknown sizes count as bytes are read or skipped.
 // MaxEntries counts visible entries; nested containers
 // have a separate MaxEntries budget. Hidden TAR extension headers are excluded.
-// Any iteration error is terminal.
+// Size and entry-count limit errors return the rejected entry alongside the error.
+// Errors remain terminal unless explicitly cleared by Recover or Skip.
 func (s *Stream) Next() (*StreamEntry, error) {
 	if s.closed {
 		return nil, fs.ErrClosed
@@ -206,20 +210,27 @@ func (s *Stream) Next() (*StreamEntry, error) {
 			return nil, err
 		}
 	}
+	s.pending = nil
+	s.reserved = 0
+	s.entryRead = false
 	entry, err := s.source.Next()
 	if err == nil {
+		s.pending = entry
 		err = s.checkEntry(entry)
+		if !errors.Is(err, ErrEntryLimit) {
+			s.entries++
+		}
 	}
 	if err != nil {
 		s.err = err
-		return nil, err
+		return entry, err
 	}
-	s.entries++
 	s.unknownSize = entry.Size == -1
 	s.entryBytes = 0
 	s.entryPath = entry.Path
 	if !s.unknownSize {
 		s.expanded += entry.Size
+		s.reserved = entry.Size
 	}
 	return entry, nil
 }
@@ -233,7 +244,7 @@ func (s *Stream) checkEntry(entry *StreamEntry) error {
 		return fmt.Errorf("%w: %s exceeds %d bytes", ErrEntrySizeLimit, entry.Path, s.options.MaxEntryBytes)
 	}
 	if entry.Size > s.options.MaxExpandedBytes-s.expanded {
-		return fmt.Errorf("%w: exceeds %d bytes", ErrDecompressLimit, s.options.MaxExpandedBytes)
+		return fmt.Errorf("%w: %s exceeds remaining budget of %d bytes", ErrDecompressLimit, entry.Path, s.options.MaxExpandedBytes-s.expanded)
 	}
 	return nil
 }
@@ -253,6 +264,9 @@ func (s *Stream) Read(p []byte) (int, error) {
 	}
 	var n int
 	var err error
+	if len(p) > 0 {
+		s.entryRead = true
+	}
 	if s.unknownSize {
 		n, err = s.readUnknownSize(p)
 	} else {
@@ -312,6 +326,37 @@ func (s *Stream) Recover() error {
 	}
 	zip.recover()
 	s.err = nil
+	s.pending = nil
+	return nil
+}
+
+// Skip abandons an unread ZIP entry without decompressing or validating it.
+// It also clears size-limit errors returned with an entry by Next. The entry
+// counts toward MaxEntries, but its unread body does not consume expanded bytes.
+// Other formats, missing entries and entries already read return ErrUnsupported.
+// Cancellation, input limits and entry-count limits remain terminal.
+func (s *Stream) Skip() error {
+	if s.closed {
+		return fs.ErrClosed
+	}
+	if err := s.ctx.Err(); err != nil {
+		return err
+	}
+	if errors.Is(s.err, io.EOF) {
+		return errors.ErrUnsupported
+	}
+	if s.err != nil && !errors.Is(s.err, ErrEntrySizeLimit) && !errors.Is(s.err, ErrDecompressLimit) {
+		return s.err
+	}
+	zip, ok := s.source.(*zipStream)
+	if !ok || s.pending == nil || s.entryRead {
+		return errors.ErrUnsupported
+	}
+	zip.recover()
+	s.expanded -= s.reserved
+	s.pending = nil
+	s.reserved = 0
+	s.err = nil
 	return nil
 }
 

stream_skip_test.go

@@ -0,0 +1,171 @@
+package archives
+
+import (
+	"archive/zip"
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"io/fs"
+	"math"
+	"testing"
+)
+
+func skipZIPFixture(t *testing.T, method uint16) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	w := zip.NewWriter(&buf)
+	for _, entry := range []struct{ name, body string }{{"big", string(bytes.Repeat([]byte{'x'}, 2700))}, {"small", "valid"}, {"tail", "end"}} {
+		body, err := w.CreateHeader(&zip.FileHeader{Name: entry.name, Method: method})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := io.WriteString(body, entry.body); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	return buf.Bytes()
+}
+
+func TestStreamSkipRejectedZIPEntry(t *testing.T) {
+	for _, method := range []uint16{zip.Store, zip.Deflate} {
+		for _, options := range []StreamOptions{{MaxEntryBytes: 5}, {MaxExpandedBytes: 5}} {
+			assertSkipRejectedZIPEntry(t, method, options)
+		}
+	}
+}
+
+func assertSkipRejectedZIPEntry(t *testing.T, method uint16, options StreamOptions) {
+	t.Helper()
+	s, err := OpenStreamBytes("renamed", skipZIPFixture(t, method), options)
+	if err != nil {
+		t.Fatal(err)
+	}
+	entry, err := s.Next()
+	if entry == nil || entry.Path != "big" || (!errors.Is(err, ErrEntrySizeLimit) && !errors.Is(err, ErrDecompressLimit)) {
+		t.Fatalf("rejected entry: %+v %v", entry, err)
+	}
+	if _, again := s.Next(); again != err {
+		t.Fatal("size rejection should remain terminal until Skip")
+	}
+	if err := s.Skip(); err != nil {
+		t.Fatal(err)
+	}
+	entry, err = s.Next()
+	if err != nil || entry.Path != "small" {
+		t.Fatalf("small entry: %+v %v", entry, err)
+	}
+	body, err := io.ReadAll(s)
+	if err != nil || string(body) != "valid" {
+		t.Fatalf("small body: %q %v", body, err)
+	}
+	_, err = s.Next()
+	if options.MaxExpandedBytes > 0 && !errors.Is(err, ErrDecompressLimit) {
+		t.Fatalf("read bytes lost from budget: %v", err)
+	}
+	if err := s.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestStreamSkipUnreadZIPEntry(t *testing.T) {
+	s, err := OpenStreamBytes("renamed", skipZIPFixture(t, zip.Deflate), StreamOptions{MaxExpandedBytes: 2700})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = s.Close() }()
+	if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) {
+		t.Fatal(err)
+	}
+	entry, err := s.Next()
+	if err != nil {
+		t.Fatal(err)
+	}
+	entry.Size = math.MaxInt64
+	if err := s.Skip(); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) {
+		t.Fatal(err)
+	}
+	entry, err = s.Next()
+	if err != nil || entry.Path != "small" {
+		t.Fatalf("reserved bytes not released: %+v %v", entry, err)
+	}
+	if _, err := s.Read(make([]byte, 1)); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) {
+		t.Fatalf("partially read entry skipped: %v", err)
+	}
+	if err := consumeStream(s, true); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) {
+		t.Fatal(err)
+	}
+}
+
+func TestStreamSkipDoesNotReadRejectedBody(t *testing.T) {
+	for _, method := range []uint16{zip.Store, zip.Deflate} {
+		s, err := OpenStreamBytes("renamed", corruptZipMember(t, method), StreamOptions{MaxExpandedBytes: 5})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := s.Next(); !errors.Is(err, ErrDecompressLimit) {
+			t.Fatal(err)
+		}
+		if err := s.Skip(); err != nil {
+			t.Fatal(err)
+		}
+		entry, err := s.Next()
+		if err != nil || entry.Path != "valid" {
+			t.Fatalf("next after corrupt skipped body: %+v %v", entry, err)
+		}
+		body, err := io.ReadAll(s)
+		if err != nil || string(body) != "valid" {
+			t.Fatalf("body: %q %v", body, err)
+		}
+		if err := s.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestStreamSkipUnsupportedFormatsAndCancellation(t *testing.T) {
+	for _, name := range []string{"test.tar", "test.tar.gz", "test.conda"} {
+		s, err := OpenStreamBytes(name, streamTestArchive(t, name, tarWithPayloads(t, make([]byte, 2700))), StreamOptions{MaxEntryBytes: 10})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := s.Next(); !errors.Is(err, ErrEntrySizeLimit) {
+			t.Fatal(err)
+		}
+		if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) {
+			t.Fatalf("%s: %v", name, err)
+		}
+		_ = s.Close()
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	s, err := OpenStreamBytesContext(ctx, "renamed", skipZIPFixture(t, zip.Store), StreamOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Next(); err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	if err := s.Skip(); !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	_ = s.Close()
+	if err := s.Skip(); !errors.Is(err, fs.ErrClosed) {
+		t.Fatal(err)
+	}
+	if _, err := OpenStreamBytes("renamed", skipZIPFixture(t, zip.Store), StreamOptions{MaxEntries: 2}); !errors.Is(err, ErrEntryLimit) {
+		t.Fatalf("entry count limit bypassed: %v", err)
+	}
+}