chore(net,extensions): wire up IPLD extensions, expose as Node instead of []byte

* Extensions now working with new dag-cbor network protocol
* dag-cbor network protocol still not default, most tests are still exercising
  the existing v1 protocol
* Metadata now using bindnode instead of cbor-gen
* []byte for deferred extensions decoding is now replaced with datamodel.Node
  everywhere. Internal extensions now using some form of go-ipld-prime
	decode to convert them to local types (metadata using bindnode, others using
	direct inspection).
* V1 protocol also using dag-cbor decode of extensions data and exporting the
  bytes - this may be a breaking change for exising extensions - need to check
	whether this should be done differently. Maybe a try-decode and if it fails
	export a wrapped Bytes Node?

Author: rvagg

cidset/cidset.go

@@ -4,32 +4,30 @@ import (
 	"errors"
 
 	"github.com/ipfs/go-cid"
+	"github.com/ipld/go-ipld-prime/datamodel"
 	"github.com/ipld/go-ipld-prime/fluent"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	"github.com/ipld/go-ipld-prime/node/basicnode"
-
-	"github.com/ipfs/go-graphsync/ipldutil"
 )
 
 // EncodeCidSet encodes a cid set into bytes for the do-no-send-cids extension
-func EncodeCidSet(cids *cid.Set) ([]byte, error) {
+func EncodeCidSet(cids *cid.Set) datamodel.Node {
 	list := fluent.MustBuildList(basicnode.Prototype.List, int64(cids.Len()), func(la fluent.ListAssembler) {
 		_ = cids.ForEach(func(c cid.Cid) error {
 			la.AssembleValue().AssignLink(cidlink.Link{Cid: c})
 			return nil
 		})
 	})
-	return ipldutil.EncodeNode(list)
+	return list
 }
 
 // DecodeCidSet decode a cid set from data for the do-no-send-cids extension
-func DecodeCidSet(data []byte) (*cid.Set, error) {
-	list, err := ipldutil.DecodeNode(data)
-	if err != nil {
-		return nil, err
+func DecodeCidSet(data datamodel.Node) (*cid.Set, error) {
+	if data.Kind() != datamodel.Kind_List {
+		return nil, errors.New("did not receive a list of CIDs")
 	}
 	set := cid.NewSet()
-	iter := list.ListIterator()
+	iter := data.ListIterator()
 	for !iter.Done() {
 		_, next, err := iter.Next()
 		if err != nil {

cidset/cidset_test.go

@@ -15,8 +15,7 @@ func TestDecodeEncodeCidSet(t *testing.T) {
 	for _, c := range cids {
 		set.Add(c)
 	}
-	encoded, err := EncodeCidSet(set)
-	require.NoError(t, err, "encode errored")
+	encoded := EncodeCidSet(set)
 	decodedCidSet, err := DecodeCidSet(encoded)
 	require.NoError(t, err, "decode errored")
 	require.Equal(t, decodedCidSet.Len(), set.Len())

dedupkey/dedupkey.go

@@ -1,27 +1,21 @@
 package dedupkey
 
 import (
+	"github.com/ipld/go-ipld-prime/datamodel"
 	"github.com/ipld/go-ipld-prime/node/basicnode"
-
-	"github.com/ipfs/go-graphsync/ipldutil"
 )
 
 // EncodeDedupKey returns encoded cbor data for string key
-func EncodeDedupKey(key string) ([]byte, error) {
+func EncodeDedupKey(key string) (datamodel.Node, error) {
 	nb := basicnode.Prototype.String.NewBuilder()
 	err := nb.AssignString(key)
 	if err != nil {
 		return nil, err
 	}
-	nd := nb.Build()
-	return ipldutil.EncodeNode(nd)
+	return nb.Build(), nil
 }
 
 // DecodeDedupKey returns a string key decoded from cbor data
-func DecodeDedupKey(data []byte) (string, error) {
-	nd, err := ipldutil.DecodeNode(data)
-	if err != nil {
-		return "", err
-	}
-	return nd.AsString()
+func DecodeDedupKey(data datamodel.Node) (string, error) {
+	return data.AsString()
 }

donotsendfirstblocks/donotsendfirstblocks.go

@@ -1,23 +1,17 @@
 package donotsendfirstblocks
 
 import (
+	"github.com/ipld/go-ipld-prime/datamodel"
 	"github.com/ipld/go-ipld-prime/node/basicnode"
-
-	"github.com/ipfs/go-graphsync/ipldutil"
 )
 
 // EncodeDoNotSendFirstBlocks returns encoded cbor data for the given number
 // of blocks to skip
-func EncodeDoNotSendFirstBlocks(skipBlockCount int64) ([]byte, error) {
-	nd := basicnode.NewInt(skipBlockCount)
-	return ipldutil.EncodeNode(nd)
+func EncodeDoNotSendFirstBlocks(skipBlockCount int64) datamodel.Node {
+	return basicnode.NewInt(skipBlockCount)
 }
 
 // DecodeDoNotSendFirstBlocks returns the number of blocks to skip
-func DecodeDoNotSendFirstBlocks(data []byte) (int64, error) {
-	nd, err := ipldutil.DecodeNode(data)
-	if err != nil {
-		return 0, err
-	}
-	return nd.AsInt()
+func DecodeDoNotSendFirstBlocks(data datamodel.Node) (int64, error) {
+	return data.AsInt()
 }

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/ipfs/go-unixfs v0.3.1
 	github.com/ipfs/go-unixfsnode v1.2.0
 	github.com/ipld/go-codec-dagpb v1.3.0
-	github.com/ipld/go-ipld-prime v0.14.4
+	github.com/ipld/go-ipld-prime v0.14.5-0.20220121142026-257b06219831
 	github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c
 	github.com/libp2p/go-buffer-pool v0.0.2
 	github.com/libp2p/go-libp2p v0.16.0
@@ -49,5 +49,3 @@ require (
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1
 	google.golang.org/protobuf v1.27.1
 )
-
-replace github.com/ipld/go-ipld-prime => ../../ipld/go-ipld-prime

go.sum

@@ -454,8 +454,9 @@ github.com/ipld/go-codec-dagpb v1.3.0/go.mod h1:ga4JTU3abYApDC3pZ00BC2RSvC3qfBb9
 github.com/ipld/go-ipld-prime v0.9.1-0.20210324083106-dc342a9917db/go.mod h1:KvBLMr4PX1gWptgkzRjVZCrLmSGcZCb/jioOQwCqZN8=
 github.com/ipld/go-ipld-prime v0.11.0/go.mod h1:+WIAkokurHmZ/KwzDOMUuoeJgaRQktHtEaLglS3ZeV8=
 github.com/ipld/go-ipld-prime v0.14.0/go.mod h1:9ASQLwUFLptCov6lIYc70GRB4V7UTyLD0IJtrDJe6ZM=
-github.com/ipld/go-ipld-prime v0.14.4 h1:bqhmume8+nbNsX4/+J6eohktfZHAI8GKrF3rQ0xgOyc=
 github.com/ipld/go-ipld-prime v0.14.4/go.mod h1:QcE4Y9n/ZZr8Ijg5bGPT0GqYWgZ1704nH0RDcQtgTP0=
+github.com/ipld/go-ipld-prime v0.14.5-0.20220121142026-257b06219831 h1:hHLYeedwqakiOMaGI6HWF84geJu2VL6OZ1DrrhyY70s=
+github.com/ipld/go-ipld-prime v0.14.5-0.20220121142026-257b06219831/go.mod h1:QcE4Y9n/ZZr8Ijg5bGPT0GqYWgZ1704nH0RDcQtgTP0=
 github.com/ipld/go-ipld-prime/storage/bsadapter v0.0.0-20211210234204-ce2a1c70cd73/go.mod h1:2PJ0JgxyB08t0b2WKrcuqI3di0V+5n6RS/LTUJhkoxY=
 github.com/jackpal/gateway v1.0.5/go.mod h1:lTpwd4ACLXmpyiCTRtfiNyVnUmqT9RivzCDQetPfnjA=
 github.com/jackpal/go-nat-pmp v1.0.1/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=

graphsync.go

@@ -8,6 +8,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/datamodel"
 	"github.com/ipld/go-ipld-prime/traversal"
 	"github.com/libp2p/go-libp2p-core/peer"
 )
@@ -54,7 +55,7 @@ type ExtensionName string
 // ExtensionData is a name/data pair for a graphsync extension
 type ExtensionData struct {
 	Name ExtensionName
-	Data []byte
+	Data datamodel.Node
 }
 
 const (
@@ -172,7 +173,7 @@ type RequestData interface {
 
 	// Extension returns the content for an extension on a response, or errors
 	// if extension is not present
-	Extension(name ExtensionName) ([]byte, bool)
+	Extension(name ExtensionName) (datamodel.Node, bool)
 
 	// IsCancel returns true if this particular request is being cancelled
 	IsCancel() bool
@@ -188,7 +189,7 @@ type ResponseData interface {
 
 	// Extension returns the content for an extension on a response, or errors
 	// if extension is not present
-	Extension(name ExtensionName) ([]byte, bool)
+	Extension(name ExtensionName) (datamodel.Node, bool)
 }
 
 // BlockData gives information about a block included in a graphsync response

impl/graphsync_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/ipfs/go-unixfsnode"
 	unixfsbuilder "github.com/ipfs/go-unixfsnode/data/builder"
 	ipld "github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/datamodel"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	"github.com/ipld/go-ipld-prime/node/basicnode"
 	"github.com/ipld/go-ipld-prime/traversal/selector"
@@ -137,7 +138,7 @@ func TestSendResponseToIncomingRequest(t *testing.T) {
 	}
 	td.gsnet1.SetDelegate(r)
 
-	var receivedRequestData []byte
+	var receivedRequestData datamodel.Node
 	// initialize graphsync on second node to response to requests
 	gsnet := td.GraphSyncHost2()
 	gsnet.RegisterIncomingRequestHook(
@@ -164,7 +165,7 @@ func TestSendResponseToIncomingRequest(t *testing.T) {
 	// read the values sent back to requestor
 	var received gsmsg.GraphSyncMessage
 	var receivedBlocks []blocks.Block
-	var receivedExtensions [][]byte
+	var receivedExtensions []datamodel.Node
 	for {
 		var message receivedMessage
 		testutil.AssertReceive(ctx, t, r.messageReceived, &message, "did not receive complete response")
@@ -355,8 +356,8 @@ func TestGraphsyncRoundTrip(t *testing.T) {
 			responder := td.GraphSyncHost2()
 			assertComplete := assertCompletionFunction(responder, 1)
 
-			var receivedResponseData []byte
-			var receivedRequestData []byte
+			var receivedResponseData datamodel.Node
+			var receivedRequestData datamodel.Node
 
 			requestor.RegisterIncomingResponseHook(
 				func(p peer.ID, responseData graphsync.ResponseData, hookActions graphsync.IncomingResponseHookActions) {
@@ -537,8 +538,7 @@ func TestGraphsyncRoundTripIgnoreCids(t *testing.T) {
 		td.blockStore1[cidlink.Link{Cid: blk.Cid()}] = blk.RawData()
 		set.Add(blk.Cid())
 	}
-	encodedCidSet, err := cidset.EncodeCidSet(set)
-	require.NoError(t, err)
+	encodedCidSet := cidset.EncodeCidSet(set)
 	extension := graphsync.ExtensionData{
 		Name: graphsync.ExtensionDoNotSendCIDs,
 		Data: encodedCidSet,
@@ -609,8 +609,7 @@ func TestGraphsyncRoundTripIgnoreNBlocks(t *testing.T) {
 		td.blockStore1[cidlink.Link{Cid: blk.Cid()}] = blk.RawData()
 	}
 
-	doNotSendFirstBlocksData, err := donotsendfirstblocks.EncodeDoNotSendFirstBlocks(50)
-	require.NoError(t, err)
+	doNotSendFirstBlocksData := donotsendfirstblocks.EncodeDoNotSendFirstBlocks(50)
 	extension := graphsync.ExtensionData{
 		Name: graphsync.ExtensionsDoNotSendFirstBlocks,
 		Data: doNotSendFirstBlocksData,
@@ -844,8 +843,8 @@ func TestPauseResumeViaUpdate(t *testing.T) {
 	defer cancel()
 	td := newGsTestData(ctx, t)
 
-	var receivedReponseData []byte
-	var receivedUpdateData []byte
+	var receivedReponseData datamodel.Node
+	var receivedUpdateData datamodel.Node
 	// initialize graphsync on first node to make requests
 	requestor := td.GraphSyncHost1()
 	assertAllResponsesReceived := assertAllResponsesReceivedFunction(requestor)
@@ -945,8 +944,8 @@ func TestPauseResumeViaUpdateOnBlockHook(t *testing.T) {
 	defer cancel()
 	td := newGsTestData(ctx, t)
 
-	var receivedReponseData []byte
-	var receivedUpdateData []byte
+	var receivedReponseData datamodel.Node
+	var receivedUpdateData datamodel.Node
 	// initialize graphsync on first node to make requests
 	requestor := td.GraphSyncHost1()
 
@@ -1639,8 +1638,8 @@ func TestGraphsyncBlockListeners(t *testing.T) {
 		blocksOutgoing++
 	})
 
-	var receivedResponseData []byte
-	var receivedRequestData []byte
+	var receivedResponseData datamodel.Node
+	var receivedRequestData datamodel.Node
 
 	requestor.RegisterIncomingResponseHook(
 		func(p peer.ID, responseData graphsync.ResponseData, hookActions graphsync.IncomingResponseHookActions) {
@@ -1718,12 +1717,12 @@ type gsTestData struct {
 	gsnet2                     gsnet.GraphSyncNetwork
 	blockStore1, blockStore2   map[ipld.Link][]byte
 	persistence1, persistence2 ipld.LinkSystem
-	extensionData              []byte
+	extensionData              datamodel.Node
 	extensionName              graphsync.ExtensionName
 	extension                  graphsync.ExtensionData
-	extensionResponseData      []byte
+	extensionResponseData      datamodel.Node
 	extensionResponse          graphsync.ExtensionData
-	extensionUpdateData        []byte
+	extensionUpdateData        datamodel.Node
 	extensionUpdate            graphsync.ExtensionData
 }
 
@@ -1814,18 +1813,18 @@ func newOptionalGsTestData(ctx context.Context, t *testing.T, network1Protocols
 	td.blockStore2 = make(map[ipld.Link][]byte)
 	td.persistence2 = testutil.NewTestStore(td.blockStore2)
 	// setup extension handlers
-	td.extensionData = testutil.RandomBytes(100)
+	td.extensionData = basicnode.NewBytes(testutil.RandomBytes(100))
 	td.extensionName = graphsync.ExtensionName("AppleSauce/McGee")
 	td.extension = graphsync.ExtensionData{
 		Name: td.extensionName,
 		Data: td.extensionData,
 	}
-	td.extensionResponseData = testutil.RandomBytes(100)
+	td.extensionResponseData = basicnode.NewBytes(testutil.RandomBytes(100))
 	td.extensionResponse = graphsync.ExtensionData{
 		Name: td.extensionName,
 		Data: td.extensionResponseData,
 	}
-	td.extensionUpdateData = testutil.RandomBytes(100)
+	td.extensionUpdateData = basicnode.NewBytes(testutil.RandomBytes(100))
 	td.extensionUpdate = graphsync.ExtensionData{
 		Name: td.extensionName,
 		Data: td.extensionUpdateData,

message/bench_test.go

@@ -22,10 +22,12 @@ func BenchmarkMessageEncodingRoundtrip(b *testing.B) {
 	root := testutil.GenerateCids(1)[0]
 	ssb := builder.NewSelectorSpecBuilder(basicnode.Prototype.Any)
 	selector := ssb.Matcher().Node()
+	bb := basicnode.Prototype.Bytes.NewBuilder()
+	bb.AssignBytes(testutil.RandomBytes(100))
 	extensionName := graphsync.ExtensionName("graphsync/awesome")
 	extension := graphsync.ExtensionData{
 		Name: extensionName,
-		Data: testutil.RandomBytes(100),
+		Data: bb.Build(),
 	}
 	id := graphsync.NewRequestID()
 	priority := graphsync.Priority(rand.Int31())

message/builder.go

@@ -111,10 +111,7 @@ func (b *Builder) ScrubResponses(requestIDs []graphsync.RequestID) uint64 {
 func (b *Builder) Build() (GraphSyncMessage, error) {
 	responses := make(map[graphsync.RequestID]GraphSyncResponse, len(b.outgoingResponses))
 	for requestID, linkMap := range b.outgoingResponses {
-		mdRaw, err := metadata.EncodeMetadata(linkMap)
-		if err != nil {
-			return GraphSyncMessage{}, err
-		}
+		mdRaw := metadata.EncodeMetadata(linkMap)
 		b.extensions[requestID] = append(b.extensions[requestID], graphsync.ExtensionData{
 			Name: graphsync.ExtensionMetadata,
 			Data: mdRaw,