feat: delegation chain validation

.gitignore

@@ -0,0 +1 @@
+*.out

client/connection.go

@@ -4,10 +4,10 @@ import (
 	"crypto/sha256"
 	"fmt"
 	"hash"
+	"iter"
 
 	"github.com/storacha-network/go-ucanto/core/invocation"
 	"github.com/storacha-network/go-ucanto/core/ipld/block"
-	"github.com/storacha-network/go-ucanto/core/iterable"
 	"github.com/storacha-network/go-ucanto/core/message"
 	"github.com/storacha-network/go-ucanto/transport"
 	"github.com/storacha-network/go-ucanto/transport/car"
@@ -96,7 +96,7 @@ func (c *conn) Hasher() hash.Hash {
 type ExecutionResponse interface {
 	// Blocks returns an iterator of all the IPLD blocks that are included in
 	// the response.
-	Blocks() iterable.Iterator[block.Block]
+	Blocks() iter.Seq2[block.Block, error]
 	// Get returns a link to a receipt, given an invocation link.
 	Get(inv ucan.Link) (ucan.Link, bool)
 }

core/car/car.go

@@ -4,22 +4,23 @@ import (
 	"bufio"
 	"fmt"
 	"io"
+	"iter"
 
 	"github.com/ipfs/go-cid"
 	cbor "github.com/ipfs/go-ipld-cbor"
 	ipldcar "github.com/ipld/go-car"
 	"github.com/ipld/go-car/util"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/multiformats/go-varint"
 	"github.com/storacha-network/go-ucanto/core/ipld"
 	"github.com/storacha-network/go-ucanto/core/ipld/block"
-	"github.com/storacha-network/go-ucanto/core/iterable"
 )
 
 // ContentType is the value the HTTP Content-Type header should have for CARs.
 // See https://www.iana.org/assignments/media-types/application/vnd.ipld.car
 const ContentType = "application/vnd.ipld.car"
 
-func Encode(roots []ipld.Link, blocks iterable.Iterator[ipld.Block]) io.Reader {
+func Encode(roots []ipld.Link, blocks iter.Seq2[ipld.Block, error]) io.Reader {
 	reader, writer := io.Pipe()
 	go func() {
 		cids := []cid.Cid{}
@@ -41,23 +42,43 @@ func Encode(roots []ipld.Link, blocks iterable.Iterator[ipld.Block]) io.Reader {
 			return
 		}
 		util.LdWrite(writer, hb)
-		for {
-			block, err := blocks.Next()
+		for block, err := range blocks {
+			if err != nil {
+				writer.CloseWithError(fmt.Errorf("writing CAR blocks: %s", err))
+				return
+			}
+			err = util.LdWrite(writer, []byte(block.Link().Binary()), block.Bytes())
 			if err != nil {
-				if err == io.EOF {
-					break
-				}
 				writer.CloseWithError(fmt.Errorf("writing CAR blocks: %s", err))
 				return
 			}
-			util.LdWrite(writer, []byte(block.Link().Binary()), block.Bytes())
 		}
 		writer.Close()
 	}()
 	return reader
 }
 
-func Decode(reader io.Reader) ([]ipld.Link, iterable.Iterator[ipld.Block], error) {
+type CarBlock interface {
+	ipld.Block
+	Offset() uint64
+	Length() uint64
+}
+
+type carBlock struct {
+	ipld.Block
+	offset uint64
+	length uint64
+}
+
+func (cb carBlock) Offset() uint64 {
+	return cb.offset
+}
+
+func (cb carBlock) Length() uint64 {
+	return cb.length
+}
+
+func Decode(reader io.Reader) ([]ipld.Link, iter.Seq2[ipld.Block, error], error) {
 	br := bufio.NewReader(reader)
 
 	h, err := ipldcar.ReadHeader(br)
@@ -69,29 +90,52 @@ func Decode(reader io.Reader) ([]ipld.Link, iterable.Iterator[ipld.Block], error
 		return nil, nil, fmt.Errorf("invalid car version: %d", h.Version)
 	}
 
+	offset, err := ipldcar.HeaderSize(h)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	var roots []ipld.Link
 	for _, r := range h.Roots {
 		roots = append(roots, cidlink.Link{Cid: r})
 	}
 
-	return roots, iterable.NewIterator(func() (ipld.Block, error) {
-		cid, bytes, err := util.ReadNode(br)
-		if err != nil {
+	r := &blkReader{br, offset}
+	return roots, func(yield func(ipld.Block, error) bool) {
+		for {
+			blk, err := r.next()
 			if err == io.EOF {
-				br = nil
+				return
+			}
+			if !yield(blk, err) {
+				return
 			}
-			return nil, err
 		}
+	}, nil
+}
 
-		hashed, err := cid.Prefix().Sum(bytes)
-		if err != nil {
-			return nil, err
-		}
+type blkReader struct {
+	br     *bufio.Reader
+	offset uint64
+}
 
-		if !hashed.Equals(cid) {
-			return nil, fmt.Errorf("mismatch in content integrity, name: %s, data: %s", cid, hashed)
-		}
+func (r *blkReader) next() (CarBlock, error) {
+	cid, bytes, err := util.ReadNode(r.br)
+	if err != nil {
+		return nil, err
+	}
+
+	hashed, err := cid.Prefix().Sum(bytes)
+	if err != nil {
+		return nil, err
+	}
+
+	if !hashed.Equals(cid) {
+		return nil, fmt.Errorf("mismatch in content integrity, name: %s, data: %s", cid, hashed)
+	}
+
+	ss := uint64(cid.ByteLen()) + uint64(len(bytes))
+	r.offset += uint64(varint.UvarintSize(ss)) + ss
 
-		return block.NewBlock(cidlink.Link{Cid: cid}, bytes), nil
-	}), nil
+	return carBlock{block.NewBlock(cidlink.Link{Cid: cid}, bytes), r.offset - uint64(len(bytes)), uint64(len(bytes))}, nil
 }

core/car/car_test.go

@@ -47,16 +47,16 @@ func TestDecodeCAR(t *testing.T) {
 		t.Fatalf("unexpected root: %s, expected: %s", roots[0], fixtures[0].root)
 	}
 
-	var blks []ipld.Block
-	for {
-		b, err := blocks.Next()
+	var blks []CarBlock
+	for b, err := range blocks {
 		if err != nil {
-			if err == io.EOF {
-				break
-			}
 			t.Fatalf("reading blocks: %s", err)
 		}
-		blks = append(blks, b)
+		cb, ok := b.(CarBlock)
+		if !ok {
+			t.Fatalf("should have returned a car block")
+		}
+		blks = append(blks, cb)
 	}
 
 	if len(blks) != len(fixtures[0].blocks) {
@@ -66,6 +66,22 @@ func TestDecodeCAR(t *testing.T) {
 		if b.String() != blks[i].Link().String() {
 			t.Fatalf("unexpected block: %s, expected: %s", b, blks[i].Link())
 		}
+		// verify offset and length can be used to directly read the block in the CAR file
+		file.Seek(int64(blks[i].Offset()), io.SeekStart)
+		data := make([]byte, blks[i].Length())
+		_, err := file.Read(data)
+		if err != nil {
+			t.Fatalf("error reading block from raw file")
+		}
+		hashed, err := blks[i].Link().(cidlink.Link).Cid.Prefix().Sum(data)
+		if err != nil {
+			t.Fatalf("error hashing block from raw file")
+		}
+
+		if hashed.String() != blks[i].Link().String() {
+			t.Fatalf("raw read from offset block: %s, expected: %s", hashed, blks[i].Link())
+		}
+
 	}
 }
 

core/dag/blockstore/blockstore.go

@@ -2,16 +2,15 @@ package blockstore
 
 import (
 	"fmt"
-	"io"
+	"iter"
 	"sync"
 
 	"github.com/storacha-network/go-ucanto/core/ipld"
-	"github.com/storacha-network/go-ucanto/core/iterable"
 )
 
 type BlockReader interface {
 	Get(link ipld.Link) (ipld.Block, bool, error)
-	Iterator() iterable.Iterator[ipld.Block]
+	Iterator() iter.Seq2[ipld.Block, error]
 }
 
 type BlockWriter interface {
@@ -33,20 +32,19 @@ func (br *blockreader) Get(link ipld.Link) (ipld.Block, bool, error) {
 	return b, ok, nil
 }
 
-func (br *blockreader) Iterator() iterable.Iterator[ipld.Block] {
-	i := 0
-	return iterable.NewIterator(func() (ipld.Block, error) {
-		if len(br.keys) <= i {
-			return nil, io.EOF
-		}
-		k := br.keys[i]
-		v, ok := br.blks[k]
-		if !ok {
-			return nil, fmt.Errorf("missing block for key: %s", k)
+func (br *blockreader) Iterator() iter.Seq2[ipld.Block, error] {
+	return func(yield func(ipld.Block, error) bool) {
+		for _, k := range br.keys {
+			v, ok := br.blks[k]
+			var err error
+			if !ok {
+				err = fmt.Errorf("missing block for key: %s", k)
+			}
+			if !yield(v, err) {
+				return
+			}
 		}
-		i++
-		return v, nil
-	})
+	}
 }
 
 type blockstore struct {
@@ -75,31 +73,29 @@ func (bs *blockstore) Get(link ipld.Link) (ipld.Block, bool, error) {
 	return bs.blockreader.Get(link)
 }
 
-func (bs *blockstore) Iterator() iterable.Iterator[ipld.Block] {
+func (bs *blockstore) Iterator() iter.Seq2[ipld.Block, error] {
 	bs.Lock()
 	defer bs.Unlock()
-	keys := bs.keys[:]
-	i := 0
-	return iterable.NewIterator(func() (ipld.Block, error) {
-		if len(keys) <= i {
-			return nil, io.EOF
-		}
-		k := keys[i]
-		v, ok := bs.blks[k]
-		if !ok {
-			return nil, fmt.Errorf("missing block for key: %s", k)
+	return func(yield func(ipld.Block, error) bool) {
+		for _, k := range bs.keys {
+			v, ok := bs.blks[k]
+			var err error
+			if !ok {
+				err = fmt.Errorf("missing block for key: %s", k)
+			}
+			if !yield(v, err) {
+				return
+			}
 		}
-		i++
-		return v, nil
-	})
+	}
 }
 
 // Option is an option configuring a block reader/writer.
 type Option func(cfg *bsConfig) error
 
 type bsConfig struct {
 	blks     []ipld.Block
-	blksiter iterable.Iterator[ipld.Block]
+	blksiter iter.Seq2[ipld.Block, error]
 }
 
 // WithBlocks configures the blocks the blockstore should contain.
@@ -111,7 +107,7 @@ func WithBlocks(blks []ipld.Block) Option {
 }
 
 // WithBlocksIterator configures the blocks the blockstore should contain.
-func WithBlocksIterator(blks iterable.Iterator[ipld.Block]) Option {
+func WithBlocksIterator(blks iter.Seq2[ipld.Block, error]) Option {
 	return func(cfg *bsConfig) error {
 		cfg.blksiter = blks
 		return nil
@@ -138,15 +134,11 @@ func NewBlockStore(options ...Option) (BlockStore, error) {
 		}
 	}
 	if cfg.blksiter != nil {
-		for {
-			b, err := cfg.blksiter.Next()
+		for b, err := range cfg.blksiter {
 			if err != nil {
-				if err == io.EOF {
-					break
-				}
 				return nil, err
 			}
-			err = bs.Put(b)
+			err := bs.Put(b)
 			if err != nil {
 				return nil, err
 			}
@@ -175,12 +167,8 @@ func NewBlockReader(options ...Option) (BlockReader, error) {
 		keys = append(keys, b.Link().String())
 	}
 	if cfg.blksiter != nil {
-		for {
-			b, err := cfg.blksiter.Next()
+		for b, err := range cfg.blksiter {
 			if err != nil {
-				if err == io.EOF {
-					break
-				}
 				return nil, err
 			}
 			_, ok := blks[b.Link().String()]
@@ -196,16 +184,8 @@ func NewBlockReader(options ...Option) (BlockReader, error) {
 }
 
 func WriteInto(view ipld.View, bs BlockWriter) error {
-	blks := view.Blocks()
-	for {
-		b, err := blks.Next()
-		if err != nil {
-			if err == io.EOF {
-				break
-			}
-			return fmt.Errorf("reading proof blocks: %s", err)
-		}
-		err = bs.Put(b)
+	for b := range view.Blocks() {
+		err := bs.Put(b)
 		if err != nil {
 			return fmt.Errorf("putting proof block: %s", err)
 		}