blockstore: implement UseWholeCIDs

Update the tests too, as a few expected whole CIDs.
For now, just make them use the option.

This required index.Index to gain a new method, GetAll, so that when
using whole CIDs, methods like Get can return true if any of the
matching indexed CIDs are an exact whole-CID match, and not just looking
at the first matching indexed CID.

GetAll is akin to an iteration over all matching indexed CIDs, and its
callback returns a boolean to say if the iteration should continue.
This allows stopping as soon as we're done.

We also remove Index.Get, instead replacing it with a helper called
GetFirst, which simply makes simple uses of GetAll a single line.

We remove the non-specced and unused index implementations, too.
They were left in place in case they were useful again,
but they haven't been so far, and their code is still in git.
Keeping them around just means updating more code when refactoring.

While at it, make ZeroLengthSectionAsEOF take a boolean and return an
option, just like the other boolean options, for consistency.

Fixes #130.


This commit was moved from ipld/go-car@c2e497e

Author: mvdan

ipld/car/v2/blockstore/example_test.go

@@ -20,7 +20,10 @@ func ExampleOpenReadOnly() {
 	// Note, `OpenReadOnly` accepts bot CARv1 and CARv2 formats and transparently generate index
 	// in the background if necessary.
 	// This instance sets ZeroLengthSectionAsEOF option to treat zero sized sections in file as EOF.
-	robs, err := blockstore.OpenReadOnly("../testdata/sample-v1.car", carv2.ZeroLengthSectionAsEOF)
+	robs, err := blockstore.OpenReadOnly("../testdata/sample-v1.car",
+		blockstore.UseWholeCIDs(true),
+		carv2.ZeroLengthSectionAsEOF(true),
+	)
 	if err != nil {
 		panic(err)
 	}

ipld/car/v2/blockstore/insertionindex.go

@@ -55,7 +55,7 @@ func newRecordFromCid(c cid.Cid, at uint64) recordDigest {
 		panic(err)
 	}
 
-	return recordDigest{d.Digest, index.Record{Cid: c, Idx: at}}
+	return recordDigest{d.Digest, index.Record{Cid: c, Offset: at}}
 }
 
 func (ii *insertionIndex) insertNoReplace(key cid.Cid, n uint64) {
@@ -77,7 +77,31 @@ func (ii *insertionIndex) Get(c cid.Cid) (uint64, error) {
 		return 0, errUnsupported
 	}
 
-	return r.Record.Idx, nil
+	return r.Record.Offset, nil
+}
+
+func (ii *insertionIndex) GetAll(c cid.Cid, fn func(uint64) bool) error {
+	d, err := multihash.Decode(c.Hash())
+	if err != nil {
+		return err
+	}
+	entry := recordDigest{digest: d.Digest}
+
+	any := false
+	iter := func(i llrb.Item) bool {
+		existing := i.(recordDigest)
+		if !bytes.Equal(existing.digest, entry.digest) {
+			// We've already looked at all entries with matching digests.
+			return false
+		}
+		any = true
+		return fn(existing.Record.Offset)
+	}
+	ii.items.AscendGreaterOrEqual(entry, iter)
+	if !any {
+		return index.ErrNotFound
+	}
+	return nil
 }
 
 func (ii *insertionIndex) Marshal(w io.Writer) error {
@@ -152,6 +176,10 @@ func (ii *insertionIndex) flatten() (index.Index, error) {
 	return si, nil
 }
 
+// note that hasExactCID is very similar to GetAll,
+// but it's separate as it allows us to compare Record.Cid directly,
+// whereas GetAll just provides Record.Offset.
+
 func (ii *insertionIndex) hasExactCID(c cid.Cid) bool {
 	d, err := multihash.Decode(c.Hash())
 	if err != nil {

ipld/car/v2/blockstore/readonly.go

@@ -65,7 +65,6 @@ type ReadOnly struct {
 // go-car/v2 package.
 func UseWholeCIDs(enable bool) carv2.ReadOption {
 	return func(o *carv2.ReadOptions) {
-		// TODO: update methods like Get, Has, and AllKeysChan to obey this.
 		o.BlockstoreUseWholeCIDs = enable
 	}
 }
@@ -177,68 +176,121 @@ func (b *ReadOnly) Has(key cid.Cid) (bool, error) {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
-	offset, err := b.idx.Get(key)
+	var fnFound bool
+	var fnErr error
+	err := b.idx.GetAll(key, func(offset uint64) bool {
+		uar := internalio.NewOffsetReadSeeker(b.backing, int64(offset))
+		var err error
+		_, err = varint.ReadUvarint(uar)
+		if err != nil {
+			fnErr = err
+			return false
+		}
+		_, readCid, err := cid.CidFromReader(uar)
+		if err != nil {
+			fnErr = err
+			return false
+		}
+		if b.ropts.BlockstoreUseWholeCIDs {
+			fnFound = readCid.Equals(key)
+			return !fnFound // continue looking if we haven't found it
+		} else {
+			fnFound = bytes.Equal(readCid.Hash(), key.Hash())
+			return false
+		}
+	})
 	if errors.Is(err, index.ErrNotFound) {
 		return false, nil
 	} else if err != nil {
 		return false, err
 	}
-	uar := internalio.NewOffsetReadSeeker(b.backing, int64(offset))
-	_, err = varint.ReadUvarint(uar)
-	if err != nil {
-		return false, err
-	}
-	_, c, err := cid.CidFromReader(uar)
-	if err != nil {
-		return false, err
-	}
-	return bytes.Equal(key.Hash(), c.Hash()), nil
+	return fnFound, fnErr
 }
 
 // Get gets a block corresponding to the given key.
 func (b *ReadOnly) Get(key cid.Cid) (blocks.Block, error) {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
-	offset, err := b.idx.Get(key)
-	if err != nil {
-		if err == index.ErrNotFound {
-			err = blockstore.ErrNotFound
+	var fnData []byte
+	var fnErr error
+	err := b.idx.GetAll(key, func(offset uint64) bool {
+		readCid, data, err := b.readBlock(int64(offset))
+		if err != nil {
+			fnErr = err
+			return false
 		}
+		if b.ropts.BlockstoreUseWholeCIDs {
+			if readCid.Equals(key) {
+				fnData = data
+				return false
+			} else {
+				return true // continue looking
+			}
+		} else {
+			if bytes.Equal(readCid.Hash(), key.Hash()) {
+				fnData = data
+			}
+			return false
+		}
+	})
+	if errors.Is(err, index.ErrNotFound) {
+		return nil, blockstore.ErrNotFound
+	} else if err != nil {
 		return nil, err
+	} else if fnErr != nil {
+		return nil, fnErr
 	}
-	entry, data, err := b.readBlock(int64(offset))
-	if err != nil {
-		return nil, err
-	}
-	if !bytes.Equal(key.Hash(), entry.Hash()) {
+	if fnData == nil {
 		return nil, blockstore.ErrNotFound
 	}
-	return blocks.NewBlockWithCid(data, key)
+	return blocks.NewBlockWithCid(fnData, key)
 }
 
 // GetSize gets the size of an item corresponding to the given key.
 func (b *ReadOnly) GetSize(key cid.Cid) (int, error) {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
-	idx, err := b.idx.Get(key)
-	if err != nil {
-		return -1, err
-	}
-	rdr := internalio.NewOffsetReadSeeker(b.backing, int64(idx))
-	sectionLen, err := varint.ReadUvarint(rdr)
-	if err != nil {
+	var fnSize int = -1
+	var fnErr error
+	err := b.idx.GetAll(key, func(offset uint64) bool {
+		rdr := internalio.NewOffsetReadSeeker(b.backing, int64(offset))
+		sectionLen, err := varint.ReadUvarint(rdr)
+		if err != nil {
+			fnErr = err
+			return false
+		}
+		cidLen, readCid, err := cid.CidFromReader(rdr)
+		if err != nil {
+			fnErr = err
+			return false
+		}
+		if b.ropts.BlockstoreUseWholeCIDs {
+			if readCid.Equals(key) {
+				fnSize = int(sectionLen) - cidLen
+				return false
+			} else {
+				return true // continue looking
+			}
+		} else {
+			if bytes.Equal(readCid.Hash(), key.Hash()) {
+				fnSize = int(sectionLen) - cidLen
+			}
+			return false
+		}
+	})
+	if errors.Is(err, index.ErrNotFound) {
 		return -1, blockstore.ErrNotFound
+	} else if err != nil {
+		return -1, err
+	} else if fnErr != nil {
+		return -1, fnErr
 	}
-	cidLen, readCid, err := cid.CidFromReader(rdr)
-	if err != nil {
-		return 0, err
-	}
-	if !readCid.Equals(key) {
+	if fnSize == -1 {
 		return -1, blockstore.ErrNotFound
 	}
-	return int(sectionLen) - cidLen, err
+	return fnSize, nil
 }
 
 // Put is not supported and always returns an error.
@@ -304,6 +356,11 @@ func (b *ReadOnly) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
 				return // TODO: log this error
 			}
 
+			// If we're just using multihashes, flatten to the "raw" codec.
+			if !b.ropts.BlockstoreUseWholeCIDs {
+				c = cid.NewCidV1(cid.Raw, c.Hash())
+			}
+
 			select {
 			case ch <- c:
 			case <-ctx.Done():

ipld/car/v2/blockstore/readonly_test.go

@@ -49,7 +49,9 @@ func TestReadOnly(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			subject, err := OpenReadOnly(tt.v1OrV2path)
+			subject, err := OpenReadOnly(tt.v1OrV2path,
+				UseWholeCIDs(true),
+			)
 			t.Cleanup(func() { subject.Close() })
 			require.NoError(t, err)
 

ipld/car/v2/blockstore/readwrite.go

@@ -89,7 +89,6 @@ func AllowDuplicatePuts(allow bool) carv2.WriteOption {
 // Resuming from finalized files is allowed. However, resumption will regenerate the index
 // regardless by scanning every existing block in file.
 func OpenReadWrite(path string, roots []cid.Cid, opts ...carv2.ReadWriteOption) (*ReadWrite, error) {
-	// TODO: enable deduplication by default now that resumption is automatically attempted.
 	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o666) // TODO: Should the user be able to configure FileMode permissions?
 	if err != nil {
 		return nil, fmt.Errorf("could not open read/write file: %w", err)