Implement initial read-only blockstore

v2/blockstore/readonly.go

@@ -0,0 +1,189 @@
+package blockstore
+
+import (
+	"bufio"
+	"context"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	blockstore "github.com/ipfs/go-ipfs-blockstore"
+	carv1 "github.com/ipld/go-car"
+	"github.com/ipld/go-car/util"
+	"github.com/ipld/go-car/v2/internal/index"
+	internalio "github.com/ipld/go-car/v2/internal/io"
+	"golang.org/x/exp/mmap"
+)
+
+var _ blockstore.Blockstore = (*ReadOnly)(nil)
+
+// errUnsupported is returned for unsupported operations
+var errUnsupported = errors.New("unsupported operation")
+
+// ReadOnly provides a read-only Car Block Store.
+type ReadOnly struct {
+	backing io.ReaderAt
+	idx     index.Index
+}
+
+// ReadOnlyOf opens a carbs data store from an existing reader of the base data and index
+func ReadOnlyOf(backing io.ReaderAt, index index.Index) *ReadOnly {
+	return &ReadOnly{backing, index}
+}
+
+// LoadReadOnly opens a read-only blockstore, generating an index if it does not exist
+func LoadReadOnly(path string, noPersist bool) (*ReadOnly, error) {
+	reader, err := mmap.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	idx, err := index.Restore(path)
+	if err != nil {
+		idx, err = index.GenerateIndex(reader, 0, index.IndexSorted)
+		if err != nil {
+			return nil, err
+		}
+		if !noPersist {
+			if err = index.Save(idx, path); err != nil {
+				return nil, err
+			}
+		}
+	}
+	obj := ReadOnly{
+		backing: reader,
+		idx:     idx,
+	}
+	return &obj, nil
+}
+
+func (b *ReadOnly) read(idx int64) (cid.Cid, []byte, error) {
+	bcid, data, err := util.ReadNode(bufio.NewReader(internalio.NewOffsetReader(b.backing, idx)))
+	return bcid, data, err
+}
+
+// DeleteBlock is unsupported and always returns an error
+func (b *ReadOnly) DeleteBlock(_ cid.Cid) error {
+	return errUnsupported
+}
+
+// Has indicates if the store has a cid
+func (b *ReadOnly) Has(key cid.Cid) (bool, error) {
+	offset, err := b.idx.Get(key)
+	if err != nil {
+		return false, err
+	}
+	uar := internalio.NewOffsetReader(b.backing, int64(offset))
+	_, err = binary.ReadUvarint(uar)
+	if err != nil {
+		return false, err
+	}
+	c, _, err := internalio.ReadCid(b.backing, uar.Offset())
+	if err != nil {
+		return false, err
+	}
+	return c.Equals(key), nil
+}
+
+// Get gets a block from the store
+func (b *ReadOnly) Get(key cid.Cid) (blocks.Block, error) {
+	offset, err := b.idx.Get(key)
+	if err != nil {
+		return nil, err
+	}
+	entry, bytes, err := b.read(int64(offset))
+	if err != nil {
+		// TODO Improve error handling; not all errors mean NotFound.
+		return nil, blockstore.ErrNotFound
+	}
+	if !entry.Equals(key) {
+		return nil, blockstore.ErrNotFound
+	}
+	return blocks.NewBlockWithCid(bytes, key)
+}
+
+// GetSize gets how big a item is
+func (b *ReadOnly) GetSize(key cid.Cid) (int, error) {
+	idx, err := b.idx.Get(key)
+	if err != nil {
+		return -1, err
+	}
+	l, err := binary.ReadUvarint(internalio.NewOffsetReader(b.backing, int64(idx)))
+	if err != nil {
+		return -1, blockstore.ErrNotFound
+	}
+	c, _, err := internalio.ReadCid(b.backing, int64(idx+l))
+	if err != nil {
+		return 0, err
+	}
+	if !c.Equals(key) {
+		return -1, blockstore.ErrNotFound
+	}
+	// get cid. validate.
+	return int(l), err
+}
+
+// Put is not supported and always returns an error
+func (b *ReadOnly) Put(blocks.Block) error {
+	return errUnsupported
+}
+
+// PutMany is not supported and always returns an error
+func (b *ReadOnly) PutMany([]blocks.Block) error {
+	return errUnsupported
+}
+
+// AllKeysChan returns the list of keys in the store
+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.
+	header, err := carv1.ReadHeader(bufio.NewReader(internalio.NewOffsetReader(b.backing, 0)))
+	if err != nil {
+		return nil, fmt.Errorf("error reading car header: %w", err)
+	}
+	offset, err := carv1.HeaderSize(header)
+	if err != nil {
+		return nil, err
+	}
+
+	ch := make(chan cid.Cid, 5)
+	go func() {
+		done := ctx.Done()
+
+		rdr := internalio.NewOffsetReader(b.backing, int64(offset))
+		for {
+			l, err := binary.ReadUvarint(rdr)
+			thisItemForNxt := rdr.Offset()
+			if err != nil {
+				return
+			}
+			c, _, err := internalio.ReadCid(b.backing, thisItemForNxt)
+			if err != nil {
+				return
+			}
+			rdr.SeekOffset(thisItemForNxt + int64(l))
+
+			select {
+			case ch <- c:
+				continue
+			case <-done:
+				return
+			}
+		}
+	}()
+	return ch, nil
+}
+
+// HashOnRead does nothing
+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.NewOffsetReader(b.backing, 0)))
+	if err != nil {
+		return nil, fmt.Errorf("error reading car header: %w", err)
+	}
+	return header.Roots, nil
+}

v2/carbs/carbs_test.go → v2/blockstore/readonly_test.go

@@ -1,6 +1,7 @@
-package carbs
+package blockstore
 
 import (
+	"github.com/ipld/go-car/v2/internal/index"
 	"os"
 	"testing"
 )
@@ -54,7 +55,7 @@ func TestIndexRT(t *testing.T) {
 	// TODO use temporary directory to run tests taht work with OS file system to avoid accidental source code modification
 	carFile := "testdata/test.car"
 
-	cf, err := Load(carFile, false)
+	cf, err := LoadReadOnly(carFile, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -71,7 +72,7 @@ func TestIndexRT(t *testing.T) {
 		t.Fatalf("failed get: %v", err)
 	}
 
-	idx, err := Restore(carFile)
+	idx, err := index.Restore(carFile)
 	if err != nil {
 		t.Fatalf("failed restore: %v", err)
 	}

v2/car.go

@@ -12,7 +12,6 @@ const (
 	HeaderSize = 40
 	// CharacteristicsSize is the fixed size of Characteristics bitfield within CAR v2 header in number of bytes.
 	CharacteristicsSize = 16
-	uint64Size          = 8
 )
 
 var (
@@ -45,17 +44,13 @@ type (
 	}
 )
 
-// WriteTo writes this characteristics to the given writer.
+// WriteTo writes this characteristics to the given w.
 func (c Characteristics) WriteTo(w io.Writer) (n int64, err error) {
-	if err = binary.Write(w, binary.LittleEndian, c.Hi); err != nil {
-		return
-	}
-	n += uint64Size
-	if err = binary.Write(w, binary.LittleEndian, c.Lo); err != nil {
-		return
-	}
-	n += uint64Size
-	return
+	buf := make([]byte, 16)
+	binary.LittleEndian.PutUint64(buf[:8], c.Hi)
+	binary.LittleEndian.PutUint64(buf[8:], c.Lo)
+	written, err := w.Write(buf)
+	return int64(written), err
 }
 
 func (c *Characteristics) ReadFrom(r io.Reader) (int64, error) {
@@ -65,8 +60,8 @@ func (c *Characteristics) ReadFrom(r io.Reader) (int64, error) {
 	if err != nil {
 		return n, err
 	}
-	c.Hi = binary.LittleEndian.Uint64(buf[:uint64Size])
-	c.Lo = binary.LittleEndian.Uint64(buf[uint64Size:])
+	c.Hi = binary.LittleEndian.Uint64(buf[:8])
+	c.Lo = binary.LittleEndian.Uint64(buf[8:])
 	return n, nil
 }
 
@@ -101,25 +96,18 @@ func (h Header) WithCarV1Padding(padding uint64) Header {
 
 // WriteTo serializes this header as bytes and writes them using the given io.Writer.
 func (h Header) WriteTo(w io.Writer) (n int64, err error) {
-	// TODO optimize write by encoding all bytes in a slice and writing once.
 	wn, err := h.Characteristics.WriteTo(w)
-	if err != nil {
-		return
-	}
 	n += wn
-	if err = binary.Write(w, binary.LittleEndian, h.CarV1Offset); err != nil {
-		return
-	}
-	n += uint64Size
-	if err = binary.Write(w, binary.LittleEndian, h.CarV1Size); err != nil {
-		return
-	}
-	n += uint64Size
-	if err = binary.Write(w, binary.LittleEndian, h.IndexOffset); err != nil {
+	if err != nil {
 		return
 	}
-	n += uint64Size
-	return
+	buf := make([]byte, 24)
+	binary.LittleEndian.PutUint64(buf[:8], h.CarV1Offset)
+	binary.LittleEndian.PutUint64(buf[8:16], h.CarV1Size)
+	binary.LittleEndian.PutUint64(buf[16:], h.IndexOffset)
+	written, err := w.Write(buf)
+	n += int64(written)
+	return n, err
 }
 
 // ReadFrom populates fields of this header from the given r.
@@ -128,16 +116,14 @@ func (h *Header) ReadFrom(r io.Reader) (int64, error) {
 	if err != nil {
 		return n, err
 	}
-	remainingSize := HeaderSize - CharacteristicsSize
-	buf := make([]byte, remainingSize)
+	buf := make([]byte, 24)
 	read, err := io.ReadFull(r, buf)
 	n += int64(read)
 	if err != nil {
 		return n, err
 	}
-	carV1RelOffset := uint64Size * 2
-	h.CarV1Offset = binary.LittleEndian.Uint64(buf[:uint64Size])
-	h.CarV1Size = binary.LittleEndian.Uint64(buf[uint64Size:carV1RelOffset])
-	h.IndexOffset = binary.LittleEndian.Uint64(buf[carV1RelOffset:])
+	h.CarV1Offset = binary.LittleEndian.Uint64(buf[:8])
+	h.CarV1Size = binary.LittleEndian.Uint64(buf[8:16])
+	h.IndexOffset = binary.LittleEndian.Uint64(buf[16:])
 	return n, nil
 }