blockstore: fast path for AllKeysChan using the index

v2/blockstore/readwrite.go

@@ -260,7 +260,38 @@ func (b *ReadWrite) Finalize() error {
 }
 
 func (b *ReadWrite) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
-	return b.ronly.AllKeysChan(ctx)
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	b.ronly.mu.Lock()
+	defer b.ronly.mu.Unlock()
+
+	if b.ronly.closed {
+		return nil, errClosed
+	}
+
+	out := make(chan cid.Cid)
+
+	go func() {
+		defer close(out)
+		err := b.idx.ForEachCid(func(c cid.Cid, _ uint64) error {
+			if !b.opts.BlockstoreUseWholeCIDs {
+				c = cid.NewCidV1(cid.Raw, c.Hash())
+			}
+			select {
+			case out <- c:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+			return nil
+		})
+		if err != nil {
+			maybeReportError(ctx, err)
+		}
+	}()
+
+	return out, nil
 }
 
 func (b *ReadWrite) Has(ctx context.Context, key cid.Cid) (bool, error) {

v2/internal/store/insertionindex.go

@@ -152,17 +152,23 @@ func (ii *InsertionIndex) Unmarshal(r io.Reader) error {
 }
 
 func (ii *InsertionIndex) ForEach(f func(multihash.Multihash, uint64) error) error {
-	var errr error
+	var err error
 	ii.items.AscendGreaterOrEqual(ii.items.Min(), func(i llrb.Item) bool {
 		r := i.(recordDigest).Record
-		err := f(r.Cid.Hash(), r.Offset)
-		if err != nil {
-			errr = err
-			return false
-		}
-		return true
+		err = f(r.Cid.Hash(), r.Offset)
+		return err == nil
+	})
+	return err
+}
+
+func (ii *InsertionIndex) ForEachCid(f func(cid.Cid, uint64) error) error {
+	var err error
+	ii.items.AscendGreaterOrEqual(ii.items.Min(), func(i llrb.Item) bool {
+		r := i.(recordDigest).Record
+		err = f(r.Cid, r.Offset)
+		return err == nil
 	})
-	return errr
+	return err
 }
 
 func (ii *InsertionIndex) Codec() multicodec.Code {