Remove dependency to bufio.Reader in internal carv1 package

Remove dependency to `bufio.Reader` in internal `carv1` package that
seems to be mainly used for peeking a byte to return appropriate error
when stream abruptly ends, relating to #36. This allows simplification
of code across the repo and remove all unnecessary wrappings of
`io.Reader` with `bufio.Reader`. This will also aid simplify the
internal IO utilities which will be done in future PRs. For now we
simply remove dependency to `bufio.Reader`

See:
- #36

v2/blockstore/readonly.go

@@ -1,7 +1,6 @@
 package blockstore
 
 import (
-	"bufio"
 	"bytes"
 	"context"
 	"errors"
@@ -128,7 +127,7 @@ func OpenReadOnly(path string) (*ReadOnly, error) {
 }
 
 func (b *ReadOnly) readBlock(idx int64) (cid.Cid, []byte, error) {
-	bcid, data, err := util.ReadNode(bufio.NewReader(internalio.NewOffsetReadSeeker(b.backing, idx)))
+	bcid, data, err := util.ReadNode(internalio.NewOffsetReadSeeker(b.backing, idx))
 	return bcid, data, err
 }
 
@@ -222,7 +221,7 @@ func (b *ReadOnly) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
 
 	// TODO we may use this walk for populating the index, and we need to be able to iterate keys in this way somewhere for index generation. In general though, when it's asked for all keys from a blockstore with an index, we should iterate through the index when possible rather than linear reads through the full car.
 	rdr := internalio.NewOffsetReadSeeker(b.backing, 0)
-	header, err := carv1.ReadHeader(bufio.NewReader(rdr))
+	header, err := carv1.ReadHeader(rdr)
 	if err != nil {
 		return nil, fmt.Errorf("error reading car header: %w", err)
 	}
@@ -281,7 +280,7 @@ func (b *ReadOnly) HashOnRead(bool) {
 
 // Roots returns the root CIDs of the backing CAR.
 func (b *ReadOnly) Roots() ([]cid.Cid, error) {
-	header, err := carv1.ReadHeader(bufio.NewReader(internalio.NewOffsetReadSeeker(b.backing, 0)))
+	header, err := carv1.ReadHeader(internalio.NewOffsetReadSeeker(b.backing, 0))
 	if err != nil {
 		return nil, fmt.Errorf("error reading car header: %w", err)
 	}

v2/blockstore/readwrite.go

@@ -1,7 +1,6 @@
 package blockstore
 
 import (
-	"bufio"
 	"context"
 	"errors"
 	"fmt"
@@ -217,7 +216,7 @@ func (b *ReadWrite) resumeWithRoots(roots []cid.Cid) error {
 
 	// Use the given CAR v1 padding to instantiate the CAR v1 reader on file.
 	v1r := internalio.NewOffsetReadSeeker(b.ReadOnly.backing, 0)
-	header, err := carv1.ReadHeader(bufio.NewReader(v1r))
+	header, err := carv1.ReadHeader(v1r)
 	if err != nil {
 		// Cannot read the CAR v1 header; the file is most likely corrupt.
 		return fmt.Errorf("error reading car header: %w", err)

v2/car_test.go

@@ -1,7 +1,6 @@
 package car_test
 
 import (
-	"bufio"
 	"bytes"
 	"testing"
 
@@ -36,7 +35,7 @@ func TestCarV2PragmaLength(t *testing.T) {
 }
 
 func TestCarV2PragmaIsValidCarV1Header(t *testing.T) {
-	v1h, err := carv1.ReadHeader(bufio.NewReader(bytes.NewReader(carv2.Pragma)))
+	v1h, err := carv1.ReadHeader(bytes.NewReader(carv2.Pragma))
 	assert.NoError(t, err, "cannot decode pragma as CBOR with CAR v1 header structure")
 	assert.Equal(t, &carv1.CarHeader{
 		Roots:   nil,

v2/index/index.go

@@ -1,12 +1,13 @@
 package index
 
 import (
-	"bufio"
 	"encoding/binary"
 	"fmt"
 	"io"
 	"os"
 
+	internalio "github.com/ipld/go-car/v2/internal/io"
+
 	"github.com/multiformats/go-multicodec"
 
 	"github.com/multiformats/go-varint"
@@ -77,8 +78,7 @@ func WriteTo(idx Index, w io.Writer) error {
 // The reader decodes the index by reading the first byte to interpret the encoding.
 // Returns error if the encoding is not known.
 func ReadFrom(r io.Reader) (Index, error) {
-	reader := bufio.NewReader(r)
-	code, err := varint.ReadUvarint(reader)
+	code, err := varint.ReadUvarint(internalio.ToByteReader(r))
 	if err != nil {
 		return nil, err
 	}
@@ -87,7 +87,7 @@ func ReadFrom(r io.Reader) (Index, error) {
 	if err != nil {
 		return nil, err
 	}
-	if err := idx.Unmarshal(reader); err != nil {
+	if err := idx.Unmarshal(r); err != nil {
 		return nil, err
 	}
 	return idx, nil

v2/index_gen.go

@@ -1,7 +1,6 @@
 package car
 
 import (
-	"bufio"
 	"fmt"
 	"io"
 	"os"
@@ -29,7 +28,7 @@ func (r readSeekerPlusByte) ReadByte() (byte, error) {
 // GenerateIndex generates index for a given car in v1 format.
 // The index can be stored using index.Save into a file or serialized using index.WriteTo.
 func GenerateIndex(v1 io.ReadSeeker) (index.Index, error) {
-	header, err := carv1.ReadHeader(bufio.NewReader(v1))
+	header, err := carv1.ReadHeader(v1)
 	if err != nil {
 		return nil, fmt.Errorf("error reading car header: %w", err)
 	}

v2/internal/carv1/car.go

@@ -1,7 +1,6 @@
 package carv1
 
 import (
-	"bufio"
 	"context"
 	"fmt"
 	"io"
@@ -57,8 +56,8 @@ func WriteCar(ctx context.Context, ds format.NodeGetter, roots []cid.Cid, w io.W
 	return nil
 }
 
-func ReadHeader(br *bufio.Reader) (*CarHeader, error) {
-	hb, err := util.LdRead(br)
+func ReadHeader(r io.Reader) (*CarHeader, error) {
+	hb, err := util.LdRead(r)
 	if err != nil {
 		return nil, err
 	}
@@ -107,13 +106,12 @@ func (cw *carWriter) writeNode(ctx context.Context, nd format.Node) error {
 }
 
 type CarReader struct {
-	br     *bufio.Reader
+	r      io.Reader
 	Header *CarHeader
 }
 
 func NewCarReader(r io.Reader) (*CarReader, error) {
-	br := bufio.NewReader(r)
-	ch, err := ReadHeader(br)
+	ch, err := ReadHeader(r)
 	if err != nil {
 		return nil, err
 	}
@@ -127,13 +125,13 @@ func NewCarReader(r io.Reader) (*CarReader, error) {
 	}
 
 	return &CarReader{
-		br:     br,
+		r:      r,
 		Header: ch,
 	}, nil
 }
 
 func (cr *CarReader) Next() (blocks.Block, error) {
-	c, data, err := util.ReadNode(cr.br)
+	c, data, err := util.ReadNode(cr.r)
 	if err != nil {
 		return nil, err
 	}

v2/internal/carv1/util/util.go

@@ -1,9 +1,10 @@
 package util
 
 import (
-	"bufio"
 	"io"
 
+	internalio "github.com/ipld/go-car/v2/internal/io"
+
 	"github.com/multiformats/go-varint"
 
 	cid "github.com/ipfs/go-cid"
@@ -14,8 +15,8 @@ type BytesReader interface {
 	io.ByteReader
 }
 
-func ReadNode(br *bufio.Reader) (cid.Cid, []byte, error) {
-	data, err := LdRead(br)
+func ReadNode(r io.Reader) (cid.Cid, []byte, error) {
+	data, err := LdRead(r)
 	if err != nil {
 		return cid.Cid{}, nil, err
 	}
@@ -60,15 +61,12 @@ func LdSize(d ...[]byte) uint64 {
 	return sum + uint64(s)
 }
 
-func LdRead(r *bufio.Reader) ([]byte, error) {
-	if _, err := r.Peek(1); err != nil { // no more blocks, likely clean io.EOF
-		return nil, err
-	}
-
-	l, err := varint.ReadUvarint(r)
+func LdRead(r io.Reader) ([]byte, error) {
+	l, err := varint.ReadUvarint(internalio.ToByteReader(r))
 	if err != nil {
-		if err == io.EOF {
-			return nil, io.ErrUnexpectedEOF // don't silently pretend this is a clean EOF
+		// If the length of bytes read is non-zero when the error is EOF then signal an unclean EOF.
+		if l > 0 && err == io.EOF {
+			return nil, io.ErrUnexpectedEOF
 		}
 		return nil, err
 	}

v2/internal/io/converter.go

@@ -0,0 +1,20 @@
+package io
+
+import "io"
+
+func ToByteReader(r io.Reader) io.ByteReader {
+	if br, ok := r.(io.ByteReader); ok {
+		return br
+	}
+	return readerPlusByte{r}
+}
+
+type readerPlusByte struct {
+	io.Reader
+}
+
+func (r readerPlusByte) ReadByte() (byte, error) {
+	var p [1]byte
+	_, err := io.ReadFull(r, p[:])
+	return p[0], err
+}

v2/reader.go

@@ -1,7 +1,6 @@
 package car
 
 import (
-	"bufio"
 	"fmt"
 	"io"
 
@@ -70,7 +69,7 @@ func (r *Reader) Roots() ([]cid.Cid, error) {
 	if r.roots != nil {
 		return r.roots, nil
 	}
-	header, err := carv1.ReadHeader(bufio.NewReader(r.CarV1Reader()))
+	header, err := carv1.ReadHeader(r.CarV1Reader())
 	if err != nil {
 		return nil, err
 	}
@@ -115,7 +114,7 @@ func (r *Reader) Close() error {
 // This function accepts both CAR v1 and v2 payloads.
 func ReadVersion(r io.Reader) (version uint64, err error) {
 	// TODO if the user provides a reader that sufficiently satisfies what carv1.ReadHeader is asking then use that instead of wrapping every time.
-	header, err := carv1.ReadHeader(bufio.NewReader(r))
+	header, err := carv1.ReadHeader(r)
 	if err != nil {
 		return
 	}